This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 30c10d58a1 Make `parquet-index` work with column paths (#10330)
30c10d58a1 is described below
commit 30c10d58a1d106b909e8c266f558b2bc52910162
Author: Eduard Karacharov <[email protected]>
AuthorDate: Thu Jul 16 13:41:02 2026 +0300
Make `parquet-index` work with column paths (#10330)
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->
--
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Current version of the tool works only with column names, so in case of
reading a file with e.g. multiple array columns, the only option it
provides is to run smth like
```sh
parquet-index myfile.parquet element
```
and this will allow to display index stats for the first array column in
the file.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
This patch includes two changes:
- Make `parquet-index` accept full column paths (e.g.
`mycolumn.list.element`), like `parquet-show-bloom-filter` tool already
does.
- Replace file offsets with decimal replesentations rather than hex
representations, as it's not memory addresses, so likely the numbers
will give more info in this case.
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
If this PR claims a performance improvement, please include evidence
such as benchmark results.
-->
Added an integration test for the `parquet-index` tool, and
`parquet-show-bloom-filter` already has a test coverage.
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
These are backwards incompatible changes for `parquet-index` tool, as
now it requires full column name instead of leaf column name (nothing
changes for simple cases when column path = column name)
---
.github/workflows/parquet.yml | 1 +
parquet/pytest/test_parquet_integration.py | 69 ++++++++++++++++++++-
parquet/src/bin/cli/mod.rs | 91 ++++++++++++++++++++++++++++
parquet/src/bin/parquet-index.rs | 17 ++++--
parquet/src/bin/parquet-show-bloom-filter.rs | 18 +++++-
5 files changed, 186 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/parquet.yml b/.github/workflows/parquet.yml
index dcc78f8be6..6a900f8f4f 100644
--- a/.github/workflows/parquet.yml
+++ b/.github/workflows/parquet.yml
@@ -178,6 +178,7 @@ jobs:
run: |
cargo install --path parquet --bin parquet-show-bloom-filter
--features=cli
cargo install --path parquet --bin parquet-fromcsv
--features=arrow,cli
+ cargo install --path parquet --bin parquet-index --features=cli
- name: Run pytest
run: |
cd parquet/pytest
diff --git a/parquet/pytest/test_parquet_integration.py
b/parquet/pytest/test_parquet_integration.py
index e7e6c64b04..4ff0163156 100755
--- a/parquet/pytest/test_parquet_integration.py
+++ b/parquet/pytest/test_parquet_integration.py
@@ -15,8 +15,16 @@
# specific language governing permissions and limitations
# under the License.
import pyspark.sql
-from pyspark.sql.types import StructType, StructField, StringType,
IntegerType, LongType
+from pyspark.sql.types import (
+ ArrayType,
+ StructType,
+ StructField,
+ StringType,
+ IntegerType,
+ LongType,
+)
import pandas as pd
+import re
from tempfile import NamedTemporaryFile, TemporaryDirectory
import subprocess
import pathlib
@@ -41,6 +49,27 @@ def create_data_and_spark_df(n):
return data, df
+def create_spark_df_with_arrays(n):
+ def make_list(x):
+ return [x for _ in range(8)]
+
+ spark = pyspark.sql.SparkSession.builder.getOrCreate()
+ data = [
+ (f"id-{i % 10}", f"name-{i%10}", make_list(i + 1), make_list(i + 2))
+ for i in range(n)
+ ]
+ schema = StructType(
+ [
+ StructField("id", StringType(), True),
+ StructField("name", StringType(), True),
+ StructField("int32_list", ArrayType(IntegerType()), True),
+ StructField("int64_list", ArrayType(LongType()), True),
+ ]
+ )
+ df = spark.createDataFrame(data, schema).repartition(1)
+ return data, df
+
+
def create_data_and_pandas_df(n):
data = [(f"id-{i % 10}", f"name-{i%10}", i * 2, i * 2 + 1) for i in
range(n)]
df = pd.DataFrame(data, columns=["id", "name", "int32", "int64"])
@@ -85,6 +114,12 @@ def get_show_filter_cli_output(output_dir, col_name,
test_values):
return subprocess.check_output(args)
+def get_index_cli_output(output_dir, col_name):
+ (parquet_file,) = sorted(pathlib.Path(output_dir).glob("*.parquet"))
+ args = ["parquet-index", parquet_file, col_name]
+ return subprocess.check_output(args)
+
+
SCHEMA = b"""message schema {
required binary id (UTF8);
required binary name (UTF8);
@@ -156,3 +191,35 @@ class TestParquetIntegration:
expected_results.append((v[2], False))
cli_output = get_show_filter_cli_output(output_dir, "int64",
test_values)
assert cli_output == get_expected_output(expected_results)
+
+
+class TestParquetPageIndexIntegration:
+ @staticmethod
+ def _assert_cli_output(out, pattern):
+ normalized = re.sub(r"\s+", " ", out.decode("utf-8")).strip()
+ if re.search(pattern, normalized):
+ return
+ raise AssertionError("Failed to find expected pattern in cli output")
+
+ def test_index(self):
+ _, df = create_spark_df_with_arrays(10)
+ with TemporaryDirectory() as output_dir:
+ df.write.parquet(output_dir, mode="overwrite")
+
+ name_index_expected_pattern = r"""
+ Page 0 at offset \d+ with length \d+ and row count 10, min
id-0, max id-9
+ """.strip()
+ cli_output = get_index_cli_output(output_dir, "id")
+ self._assert_cli_output(cli_output, name_index_expected_pattern)
+
+ int32_list_index_expected_pattern = r"""
+ Page 0 at offset \d+ with length \d+ and row count 10, min 1,
max 10
+ """.strip()
+ cli_output = get_index_cli_output(output_dir,
"int32_list.list.element")
+ self._assert_cli_output(cli_output,
int32_list_index_expected_pattern)
+
+ cli_output = get_index_cli_output(output_dir,
"int64_list.list.element")
+ int64_list_index_expected_pattern = r"""
+ Page 0 at offset \d+ with length \d+ and row count 10, min 2,
max 11
+ """.strip()
+ self._assert_cli_output(cli_output,
int64_list_index_expected_pattern)
diff --git a/parquet/src/bin/cli/mod.rs b/parquet/src/bin/cli/mod.rs
new file mode 100644
index 0000000000..e04b285acc
--- /dev/null
+++ b/parquet/src/bin/cli/mod.rs
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/// Parses a dot-separated Parquet column path, using `\` as an escape
character
+/// for literal dots and backslashes.
+pub fn parse_column_path(path: &str) -> Result<Vec<String>, String> {
+ let mut parts = Vec::new();
+ let mut current = String::new();
+ let mut escaped = false;
+
+ for c in path.chars() {
+ if escaped {
+ match c {
+ '.' | '\\' => current.push(c),
+ _ => {
+ return Err(format!(
+ "Invalid escape sequence \\{c} in column path {}",
+ path
+ ));
+ }
+ }
+ escaped = false;
+ continue;
+ }
+
+ match c {
+ '\\' => escaped = true,
+ '.' => {
+ parts.push(current);
+ current = String::new();
+ }
+ _ => current.push(c),
+ }
+ }
+
+ if escaped {
+ return Err(format!(
+ "Column path {} ends with an incomplete escape sequence",
+ path
+ ));
+ }
+
+ parts.push(current);
+ Ok(parts)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parse_column_path_splits_unescaped_dots() {
+ assert_eq!(parse_column_path("a.b.c").unwrap(), vec!["a", "b", "c"]);
+ }
+
+ #[test]
+ fn parse_column_path_allows_escaped_dots() {
+ assert_eq!(parse_column_path(r"a\.b.c").unwrap(), vec!["a.b", "c"]);
+ }
+
+ #[test]
+ fn parse_column_path_allows_escaped_backslashes() {
+ assert_eq!(parse_column_path(r"a\\b.c").unwrap(), vec![r"a\b", "c"]);
+ }
+
+ #[test]
+ fn parse_column_path_invalid_escape_sequence() {
+ let err = parse_column_path(r"a.\b.c").unwrap_err();
+ assert!(err.contains("Invalid escape sequence"));
+ }
+
+ #[test]
+ fn parse_column_path_incomplete_escape_sequence() {
+ let err = parse_column_path(r"a.b.c\").unwrap_err();
+ assert!(err.contains("incomplete escape sequence"));
+ }
+}
diff --git a/parquet/src/bin/parquet-index.rs b/parquet/src/bin/parquet-index.rs
index 397a75c76a..241fc20533 100644
--- a/parquet/src/bin/parquet-index.rs
+++ b/parquet/src/bin/parquet-index.rs
@@ -25,15 +25,17 @@
//! ```
//! After this `parquet-index` should be available:
//! ```
-//! parquet-index XYZ.parquet COLUMN_NAME
+//! parquet-index XYZ.parquet COLUMN_PATH
//! ```
//!
//! The binary can also be built from the source code and run as follows:
//! ```
-//! cargo run --features=cli --bin parquet-index XYZ.parquet COLUMN_NAME
+//! cargo run --features=cli --bin parquet-index XYZ.parquet COLUMN_PATH
//!
//! [page index]:
https://github.com/apache/parquet-format/blob/master/PageIndex.md
+mod cli;
+
use clap::Parser;
use parquet::data_type::ByteArray;
use parquet::errors::{ParquetError, Result};
@@ -51,7 +53,9 @@ struct Args {
#[clap(help("Path to a parquet file"))]
file: String,
- #[clap(help("Column name to print"))]
+ #[clap(help(
+ "Dot-separated column path to get index for. Literal dots can be
escaped as `\\.`"
+ ))]
column: String,
}
@@ -62,10 +66,11 @@ impl Args {
let reader = SerializedFileReader::new_with_options(file, options)?;
let schema = reader.metadata().file_metadata().schema_descr();
+ let column_path =
cli::parse_column_path(&self.column).map_err(ParquetError::General)?;
let column_idx = schema
.columns()
.iter()
- .position(|x| x.name() == self.column.as_str())
+ .position(|x| x.path().parts() == column_path.as_slice())
.ok_or_else(|| {
ParquetError::General(format!("Failed to find column {}",
self.column))
})?;
@@ -158,7 +163,7 @@ fn print_index<T: std::fmt::Display>(
.enumerate()
{
print!(
- "Page {:>5} at offset {:#010x} with length {:>10} and row count
{:>10}",
+ "Page {:>5} at offset {:>10} with length {:>10} and row count
{:>10}",
idx, o.offset, o.compressed_page_size, row_count
);
match min {
@@ -197,7 +202,7 @@ fn print_bytes_index(
.enumerate()
{
print!(
- "Page {:>5} at offset {:#010x} with length {:>10} and row count
{:>10}",
+ "Page {:>5} at offset {:>10} with length {:>10} and row count
{:>10}",
idx, o.offset, o.compressed_page_size, row_count
);
match min {
diff --git a/parquet/src/bin/parquet-show-bloom-filter.rs
b/parquet/src/bin/parquet-show-bloom-filter.rs
index 2b052a45f2..436436d9a2 100644
--- a/parquet/src/bin/parquet-show-bloom-filter.rs
+++ b/parquet/src/bin/parquet-show-bloom-filter.rs
@@ -33,6 +33,8 @@
//! cargo run --features=cli --bin parquet-show-bloom-filter -- --file-name
XYZ.parquet --column id --values a
//! ```
+mod cli;
+
use clap::Parser;
use parquet::basic::Type;
use parquet::bloom_filter::Sbbf;
@@ -49,8 +51,10 @@ use std::{fs::File, path::Path};
struct Args {
#[clap(help("Path to the parquet file"))]
file_name: String,
- #[clap(help(
- "Check the bloom filter indexes for the given column. Only string
typed columns or columns with an Int32 or Int64 physical type are supported"
+ #[clap(help(concat!(
+ "Dot-separated column path to check bloom filter for. Literal dots can
be escaped as `\\.`. ",
+ "Only string typed columns or columns with an Int32 or Int64 physical
type are supported."
+ )
))]
column: String,
#[clap(
@@ -80,6 +84,14 @@ fn main() {
)
.expect("Unable to open file as Parquet");
let metadata = file_reader.metadata();
+ let column_path = match cli::parse_column_path(&args.column) {
+ Ok(path) => path,
+ Err(err) => {
+ eprintln!("{}", err);
+ std::process::exit(1);
+ }
+ };
+
for (ri, row_group) in metadata.row_groups().iter().enumerate() {
println!("Row group #{ri}");
println!("{}", "=".repeat(80));
@@ -87,7 +99,7 @@ fn main() {
.columns()
.iter()
.enumerate()
- .find(|(_, column)| column.column_path().string() == args.column)
+ .find(|(_, column)| column.column_path().parts() ==
column_path.as_slice())
{
let row_group_reader = file_reader
.get_row_group(ri)
