Fix TPC-H decimal fixtures and log schemas

.github/workflows/pr.yaml

@@ -53,6 +53,8 @@ jobs:
           filters: |
             benchmark_data_tools:
               - "benchmark_data_tools/**"
+              - "common/testing/integration_tests/data/tpch/**"
+              - "presto/testing/common/schemas/tpch/**"
 
   test_benchmark_data_tools:
     needs: detect-changes

.github/workflows/presto-test.yml

@@ -206,6 +206,7 @@ jobs:
           cat "${DATA_DIR}/metadata.json"
           echo "--- generated files ---"
           find "${DATA_DIR}" -type f -exec ls -lh {} + | awk '{print $5, $NF}' | sort -k2
+          benchmark_data_tools/scripts/log_parquet_schemas.sh --data-dir-path "${DATA_DIR}"
 
       - &login_ghcr
         name: Log in to the Container registry

benchmark_data_tools/log_parquet_schemas.py

@@ -0,0 +1,48 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+# SPDX-License-Identifier: Apache-2.0
+
+import argparse
+from pathlib import Path
+
+import pyarrow.parquet as pq
+
+
+def get_schema_columns(parquet_path):
+    schema = pq.ParquetFile(parquet_path).schema
+    columns = tuple((column.path, column.physical_type, str(column.logical_type)) for column in schema)
+    return str(schema), columns
+
+
+def log_parquet_schemas(data_dir_path):
+    data_dir = Path(data_dir_path)
+    parquet_paths = sorted(data_dir.rglob("*.parquet"))
+    if not parquet_paths:
+        raise FileNotFoundError(f"No Parquet files found under '{data_dir}'")
+
+    table_schemas = {}
+    for parquet_path in parquet_paths:
+        relative_path = parquet_path.relative_to(data_dir)
+        table_name = relative_path.parts[0]
+        schema_text, columns = get_schema_columns(parquet_path)
+        if table_name in table_schemas:
+            first_path, first_schema_text, _ = table_schemas[table_name]
+            if schema_text != first_schema_text:
+                raise ValueError(
+                    f"Parquet schema mismatch for table '{table_name}': '{first_path}' differs from '{relative_path}'"
+                )
+            continue
+        table_schemas[table_name] = (relative_path, schema_text, columns)
+
+    for table_name, (relative_path, _, columns) in sorted(table_schemas.items()):
+        print(f"--- parquet schema: {table_name} ({relative_path}) ---")
+        for column_path, physical_type, logical_type in columns:
+            print(f"{column_path}: physical={physical_type}, logical={logical_type}")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Log physical and logical schemas for a Parquet dataset.")
+    parser.add_argument(
+        "--data-dir-path", type=str, required=True, help="Root directory containing table Parquet files."
+    )
+    args = parser.parse_args()
+    log_parquet_schemas(args.data_dir_path)

benchmark_data_tools/scripts/log_parquet_schemas.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+# SPDX-License-Identifier: Apache-2.0
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+BENCHMARK_DATA_TOOLS_DIR="$SCRIPT_DIR/.."
+REPO_ROOT="$BENCHMARK_DATA_TOOLS_DIR/.."
+
+cd "$BENCHMARK_DATA_TOOLS_DIR"
+"$REPO_ROOT/scripts/run_py_script.sh" -p "$BENCHMARK_DATA_TOOLS_DIR/log_parquet_schemas.py" -- "$@"

benchmark_data_tools/tests/integration_fixture_schema_test.py

@@ -0,0 +1,51 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+# SPDX-License-Identifier: Apache-2.0
+
+import re
+from pathlib import Path
+
+import pyarrow.parquet as pq
+
+EXPECTED_DECIMAL_COLUMNS = {
+    "c_acctbal",
+    "l_quantity",
+    "l_extendedprice",
+    "l_discount",
+    "l_tax",
+    "o_totalprice",
+    "p_retailprice",
+    "ps_supplycost",
+    "s_acctbal",
+}
+
+
+def test_tpch_integration_fixtures_use_decimal_types():
+    repo_root = Path(__file__).resolve().parents[2]
+    fixtures_dir = repo_root / "common/testing/integration_tests/data/tpch"
+    schemas_dir = repo_root / "presto/testing/common/schemas/tpch"
+    schema_decimal_columns = set()
+    for schema_path in schemas_dir.glob("*.sql"):
+        schema_decimal_columns.update(
+            re.findall(
+                r"^[ \t]+(\w+)[ \t]+DECIMAL[ \t]*\([ \t]*15[ \t]*,[ \t]*2[ \t]*\)",
+                schema_path.read_text(),
+                flags=re.IGNORECASE | re.MULTILINE,
+            )
+        )
+
+    assert schema_decimal_columns == EXPECTED_DECIMAL_COLUMNS
+
+    seen_decimal_columns = set()
+
+    for parquet_path in fixtures_dir.rglob("*.parquet"):
+        schema = pq.ParquetFile(parquet_path).schema
+        for column in schema:
+            if column.name not in EXPECTED_DECIMAL_COLUMNS:
+                continue
+            assert column.physical_type == "INT64", f"Unexpected physical type for {column.name} in {parquet_path}"
+            assert column.converted_type == "DECIMAL", f"Missing DECIMAL annotation for {column.name} in {parquet_path}"
+            assert column.precision == 15, f"Unexpected precision for {column.name} in {parquet_path}"
+            assert column.scale == 2, f"Unexpected scale for {column.name} in {parquet_path}"
+            seen_decimal_columns.add(column.name)
+
+    assert seen_decimal_columns == EXPECTED_DECIMAL_COLUMNS

benchmark_data_tools/tests/parquet_schema_logging_test.py

@@ -0,0 +1,41 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION.
+# SPDX-License-Identifier: Apache-2.0
+
+from decimal import Decimal
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+from log_parquet_schemas import log_parquet_schemas
+
+
+def test_log_parquet_schemas(tmp_path, capsys):
+    table_dir = tmp_path / "lineitem"
+    table_dir.mkdir()
+    table = pa.table(
+        {
+            "l_orderkey": pa.array([1], type=pa.int64()),
+            "l_extendedprice": pa.array([Decimal("10.25")], type=pa.decimal128(15, 2)),
+            "l_discount_float": pa.array([0.05], type=pa.float64()),
+        }
+    )
+    pq.write_table(table, table_dir / "lineitem-1.parquet", store_decimal_as_integer=True)
+
+    log_parquet_schemas(tmp_path)
+
+    assert capsys.readouterr().out.splitlines() == [
+        "--- parquet schema: lineitem (lineitem/lineitem-1.parquet) ---",
+        "l_orderkey: physical=INT64, logical=None",
+        "l_extendedprice: physical=INT64, logical=Decimal(precision=15, scale=2)",
+        "l_discount_float: physical=DOUBLE, logical=None",
+    ]
+
+
+def test_log_parquet_schemas_rejects_mismatched_partitions(tmp_path):
+    table_dir = tmp_path / "lineitem"
+    table_dir.mkdir()
+    pq.write_table(pa.table({"value": pa.array([1], type=pa.int64())}), table_dir / "lineitem-1.parquet")
+    pq.write_table(pa.table({"value": pa.array([1.0], type=pa.float64())}), table_dir / "lineitem-2.parquet")
+
+    with pytest.raises(ValueError, match="Parquet schema mismatch for table 'lineitem'"):
+        log_parquet_schemas(tmp_path)

common/testing/integration_tests/data/tpch/metadata.json

@@ -1,3 +1,4 @@
 {
-  "scale_factor": 0.01
+  "scale_factor": 0.01,
+  "approx_row_group_bytes": 134217728
 }

presto/testing/integration_tests/common_fixtures.py

@@ -33,7 +33,10 @@ def setup_and_teardown(request, presto_cursor):
         create_hive_tables.create_tables(presto_cursor, schema_name, schemas_dir, data_sub_directory)
 
     tables = presto_cursor.execute(f"SHOW TABLES in {schema_name}").fetchall()
-    for (table,) in tables:
+    for (table,) in sorted(tables):
+        create_table = presto_cursor.execute(f"SHOW CREATE TABLE hive.{schema_name}.{table}").fetchone()[0]
+        print(f"--- presto schema: hive.{schema_name}.{table} ---")
+        print(create_table)
         location = get_table_external_location(schema_name, table, presto_cursor)
         print(f"  {schema_name}.{table}: location={location}")
         if not request.config.getoption("--reference-results-dir"):