This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 9d75f1148 chore: add script and CI check to sync vendored DataFusion
protos (#1991)
9d75f1148 is described below
commit 9d75f11488fc7c655a1a29898acdaaaf8b5d2e7e
Author: Andy Grove <[email protected]>
AuthorDate: Sat Jul 11 07:01:29 2026 -0600
chore: add script and CI check to sync vendored DataFusion protos (#1991)
* chore: add script and CI check to sync vendored DataFusion protos
The vendored `ballista/core/proto/datafusion.proto` and
`datafusion_common.proto`
are build-time stubs so protoc can resolve `ballista.proto`'s imports; no
Rust is
generated from them (build.rs maps their packages to the datafusion-proto
crates
via extern_path). They must stay compatible with the pinned DataFusion
version but
were previously synced by hand, which drifts.
Add `dev/update_datafusion_proto.py` to copy the `.proto` files out of the
crate
source resolved by `cargo metadata --locked` (rewriting the one differing
import),
with a `--check` mode wired into CI via `ci/scripts/rust_proto_check.sh`.
Releases
that do not ship a `.proto` (datafusion-proto-common before v54) are
skipped rather
than failing. Also remind the datafusion version-bump script to re-run the
sync.
* chore: sync vendored DataFusion protos to 54.0.0
Run dev/update_datafusion_proto.py to bring the vendored proto stubs in line
with the pinned datafusion-proto / datafusion-proto-common 54.0.0 crates.
The
crates were upgraded to 54 without re-syncing these files; no Rust is
generated
from them so the drift was benign, but the new CI check now enforces parity.
---
.github/workflows/rust.yml | 15 +++
ballista/core/proto/datafusion.proto | 94 ++++++++++++++++--
ballista/core/proto/datafusion_common.proto | 50 +++++++++-
ci/scripts/rust_proto_check.sh | 24 +++++
dev/update_datafusion_proto.py | 143 ++++++++++++++++++++++++++++
dev/update_datafusion_versions.py | 8 ++
6 files changed, 324 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index cb0a3a48e..981829053 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -251,6 +251,21 @@ jobs:
exit 1
fi
+ datafusion-proto-sync-check:
+ name: Check vendored DataFusion proto is in sync
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/[email protected]
+ with:
+ fetch-depth: 1
+ - name: Setup Rust toolchain
+ run: |
+ rustup toolchain install stable
+ rustup default stable
+ - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4
#2.9.1
+ - name: Check vendored DataFusion proto matches pinned crate
+ run: ci/scripts/rust_proto_check.sh
+
# Coverage job was failing.
https://github.com/apache/arrow-datafusion/issues/590 tracks re-instating it
# coverage:
diff --git a/ballista/core/proto/datafusion.proto
b/ballista/core/proto/datafusion.proto
index fe81be0d4..69c003acb 100644
--- a/ballista/core/proto/datafusion.proto
+++ b/ballista/core/proto/datafusion.proto
@@ -62,6 +62,7 @@ message LogicalPlanNode {
RecursiveQueryNode recursive_query = 31;
CteWorkTableScanNode cte_work_table_scan = 32;
DmlNode dml = 33;
+ EmptyTableScanNode empty_table_scan = 34;
}
}
@@ -228,6 +229,7 @@ message AnalyzeNode {
message ExplainNode {
LogicalPlanNode input = 1;
bool verbose = 2;
+ datafusion_common.ExplainFormat format = 3;
}
message AggregateNode {
@@ -270,6 +272,25 @@ message CopyToNode {
repeated string partition_by = 7;
}
+// Identifies a built-in file format supported by DataFusion.
+// Used by DefaultLogicalExtensionCodec to serialize/deserialize
+// FileFormatFactory instances (e.g. in CopyTo plans).
+enum FileFormatKind {
+ FILE_FORMAT_KIND_UNSPECIFIED = 0;
+ FILE_FORMAT_KIND_CSV = 1;
+ FILE_FORMAT_KIND_JSON = 2;
+ FILE_FORMAT_KIND_PARQUET = 3;
+ FILE_FORMAT_KIND_ARROW = 4;
+ FILE_FORMAT_KIND_AVRO = 5;
+}
+
+// Wraps a serialized FileFormatFactory with its format kind tag,
+// so the decoder can dispatch to the correct format-specific codec.
+message FileFormatProto {
+ FileFormatKind kind = 1;
+ bytes encoded_file_format = 2;
+}
+
message DmlNode{
enum Type {
UPDATE = 0;
@@ -407,6 +428,8 @@ message LogicalExprNode {
Unnest unnest = 35;
+ // Subquery expressions
+ ScalarSubqueryExprNode scalar_subquery_expr = 36;
}
}
@@ -414,6 +437,15 @@ message Wildcard {
TableReference qualifier = 1;
}
+message SubqueryNode {
+ LogicalPlanNode subquery = 1;
+ repeated LogicalExprNode outer_ref_columns = 2;
+}
+
+message ScalarSubqueryExprNode {
+ SubqueryNode subquery = 1;
+}
+
message PlaceholderNode {
string id = 1;
// We serialize the data type, metadata, and nullability separately to
maintain
@@ -594,11 +626,15 @@ message WhenThen {
message CastNode {
LogicalExprNode expr = 1;
datafusion_common.ArrowType arrow_type = 2;
+ map<string, string> metadata = 3;
+ optional bool nullable = 4;
}
message TryCastNode {
LogicalExprNode expr = 1;
datafusion_common.ArrowType arrow_type = 2;
+ map<string, string> metadata = 3;
+ optional bool nullable = 4;
}
message SortExprNode {
@@ -752,6 +788,7 @@ message PhysicalPlanNode {
AsyncFuncExecNode async_func = 36;
BufferExecNode buffer = 37;
ArrowScanExecNode arrow_scan = 38;
+ ScalarSubqueryExecNode scalar_subquery = 39;
}
}
@@ -853,11 +890,9 @@ message PhysicalExprNode {
reserved 17;
// Unique identifier for this expression to do deduplication during
deserialization.
- // When serializing, this is set to a unique identifier for each combination
of
- // expression, process and serialization run.
- // When deserializing, if this ID has been seen before, the cached Arc is
returned
- // instead of creating a new one, enabling reconstruction of referential
integrity
- // across serde roundtrips.
+ // When serializing, this is set via `PhysicalExpr::expression_id`. When
deserializing,
+ // this id is used by the `DeduplicatingProtoConverter` to preserve
referential
+ // integrity across serde roundtrips for different expressions with the same
id.
optional uint64 expr_id = 30;
oneof ExprType {
@@ -897,9 +932,21 @@ message PhysicalExprNode {
UnknownColumn unknown_column = 20;
PhysicalHashExprNode hash_expr = 21;
+
+ PhysicalScalarSubqueryExprNode scalar_subquery = 22;
+
+ PhysicalDynamicFilterNode dynamic_filter = 23;
}
}
+message PhysicalDynamicFilterNode {
+ repeated PhysicalExprNode children = 1;
+ repeated PhysicalExprNode remapped_children = 2;
+ uint64 generation = 3;
+ PhysicalExprNode inner_expr = 4;
+ bool is_complete = 5;
+}
+
message PhysicalScalarUdfNode {
string name = 1;
repeated PhysicalExprNode args = 2;
@@ -958,6 +1005,9 @@ message PhysicalBinaryExprNode {
PhysicalExprNode l = 1;
PhysicalExprNode r = 2;
string op = 3;
+ // Linearized operands for chains of the same operator (e.g. a AND b AND c).
+ // When present, `l` and `r` are ignored and `operands` holds the flattened
list.
+ repeated PhysicalExprNode operands = 4;
}
message PhysicalDateTimeIntervalExprNode {
@@ -1018,9 +1068,6 @@ message PhysicalExtensionExprNode {
message PhysicalHashExprNode {
repeated PhysicalExprNode on_columns = 1;
uint64 seed0 = 2;
- uint64 seed1 = 3;
- uint64 seed2 = 4;
- uint64 seed3 = 5;
string description = 6;
}
@@ -1141,6 +1188,8 @@ message HashJoinExecNode {
JoinFilter filter = 8;
repeated uint32 projection = 9;
bool null_aware = 10;
+ // Optional dynamic filter expression for pushing down to the probe side.
+ PhysicalExprNode dynamic_filter = 11;
}
enum StreamPartitionMode {
@@ -1179,6 +1228,10 @@ message AnalyzeExecNode {
bool show_statistics = 2;
PhysicalPlanNode input = 3;
datafusion_common.Schema schema = 4;
+ // Optional metric category filter.
+ // Empty means "plan only". Absent (has_metric_categories=false) means "all".
+ bool has_metric_categories = 5;
+ repeated string metric_categories = 6;
}
message CrossJoinExecNode {
@@ -1268,6 +1321,8 @@ message AggregateExecNode {
repeated MaybeFilter filter_expr = 10;
AggLimit limit = 11;
bool has_grouping_set = 12;
+ // Optional dynamic filter expression for pushing down to the child.
+ PhysicalExprNode dynamic_filter = 13;
}
message GlobalLimitExecNode {
@@ -1289,6 +1344,8 @@ message SortExecNode {
// Maximum number of highest/lowest rows to fetch; negative means no limit
int64 fetch = 3;
bool preserve_partitioning = 4;
+ // Optional dynamic filter expression for TopK pushdown.
+ PhysicalExprNode dynamic_filter = 5;
}
message SortPreservingMergeExecNode {
@@ -1330,6 +1387,7 @@ message RepartitionExecNode{
// uint64 unknown = 4;
// }
Partitioning partitioning = 5;
+ bool preserve_order = 6;
}
message Partitioning {
@@ -1384,6 +1442,13 @@ message CteWorkTableScanNode {
datafusion_common.Schema schema = 2;
}
+message EmptyTableScanNode {
+ TableReference table_name = 1;
+ datafusion_common.Schema schema = 2;
+ ProjectionColumns projection = 3;
+ repeated LogicalExprNode filters = 4;
+}
+
enum GenerateSeriesName {
GS_GENERATE_SERIES = 0;
GS_RANGE = 1;
@@ -1449,4 +1514,15 @@ message AsyncFuncExecNode {
message BufferExecNode {
PhysicalPlanNode input = 1;
uint64 capacity = 2;
-}
\ No newline at end of file
+}
+
+message ScalarSubqueryExecNode {
+ PhysicalPlanNode input = 1;
+ repeated PhysicalPlanNode subqueries = 2;
+}
+
+message PhysicalScalarSubqueryExprNode {
+ datafusion_common.ArrowType data_type = 1;
+ bool nullable = 2;
+ uint32 index = 3;
+}
diff --git a/ballista/core/proto/datafusion_common.proto
b/ballista/core/proto/datafusion_common.proto
index 5848455cb..dd3588d13 100644
--- a/ballista/core/proto/datafusion_common.proto
+++ b/ballista/core/proto/datafusion_common.proto
@@ -199,7 +199,7 @@ message Union{
repeated int32 type_ids = 3;
}
-// Used for List/FixedSizeList/LargeList/Struct/Map
+// Used for List/FixedSizeList/LargeList/ListView/LargeListView/Struct/Map
message ScalarNestedValue {
message Dictionary {
bytes ipc_message = 1;
@@ -306,6 +306,8 @@ message ScalarValue{
ScalarNestedValue large_list_value = 16;
ScalarNestedValue list_value = 17;
ScalarNestedValue fixed_size_list_value = 18;
+ ScalarNestedValue list_view_value = 46;
+ ScalarNestedValue large_list_view_value = 47;
ScalarNestedValue struct_value = 32;
ScalarNestedValue map_value = 41;
@@ -398,6 +400,8 @@ message ArrowType{
List LIST = 25;
List LARGE_LIST = 26;
FixedSizeList FIXED_SIZE_LIST = 27;
+ List LIST_VIEW = 43;
+ List LARGE_LIST_VIEW = 44;
Struct STRUCT = 28;
Union UNION = 29;
Dictionary DICTIONARY = 30;
@@ -430,6 +434,13 @@ message JsonWriterOptions {
}
+enum CsvQuoteStyle {
+ NECESSARY = 0;
+ ALWAYS = 1;
+ NON_NUMERIC = 2;
+ NEVER = 3;
+}
+
message CsvWriterOptions {
// Compression type
CompressionTypeVariant compression = 1;
@@ -453,6 +464,12 @@ message CsvWriterOptions {
string escape = 10;
// Optional flag whether to double quotes, instead of escaping. Defaults to
`true`
bool double_quote = 11;
+ // Quote style for CSV writing
+ CsvQuoteStyle quote_style = 12;
+ // Whether to ignore leading whitespace in string values
+ bool ignore_leading_whitespace = 13;
+ // Whether to ignore trailing whitespace in string values
+ bool ignore_trailing_whitespace = 14;
}
// Options controlling CSV format
@@ -476,6 +493,12 @@ message CsvOptions {
bytes terminator = 17; // Optional terminator character as a byte
bytes truncated_rows = 18; // Indicates if truncated rows are allowed
optional uint32 compression_level = 19; // Optional compression level
+ // Quote style for CSV writing
+ CsvQuoteStyle quote_style = 20;
+ // Whether to ignore leading whitespace in string values
+ bytes ignore_leading_whitespace = 21;
+ // Whether to ignore trailing whitespace in string values
+ bytes ignore_trailing_whitespace = 22;
}
// Options controlling CSV format
@@ -603,6 +626,24 @@ message ParquetOptions {
oneof max_predicate_cache_size_opt {
uint64 max_predicate_cache_size = 33;
}
+
+ ParquetCdcOptions content_defined_chunking = 35;
+
+ // Optional timezone applied to INT96-coerced timestamps when `coerce_int96`
+ // is set. When `Some`, INT96 columns coerce to
+ // `Timestamp(<coerce_int96>, Some(<tz>))` instead of the default
+ // `Timestamp(<coerce_int96>, None)`. No effect when `coerce_int96` is unset.
+ oneof coerce_int96_tz_opt {
+ string coerce_int96_tz = 36;
+ }
+}
+
+// Content-defined chunking (CDC) options for writing parquet files.
+message ParquetCdcOptions {
+ bool enabled = 1;
+ uint64 min_chunk_size = 2;
+ uint64 max_chunk_size = 3;
+ int32 norm_level = 4;
}
enum JoinSide {
@@ -637,4 +678,11 @@ message ColumnStats {
Precision null_count = 3;
Precision distinct_count = 4;
Precision byte_size = 6;
+}
+
+enum ExplainFormat {
+ EXPLAIN_FORMAT_INDENT = 0;
+ EXPLAIN_FORMAT_TREE = 1;
+ EXPLAIN_FORMAT_PGJSON = 2;
+ EXPLAIN_FORMAT_GRAPHVIZ = 3;
}
\ No newline at end of file
diff --git a/ci/scripts/rust_proto_check.sh b/ci/scripts/rust_proto_check.sh
new file mode 100755
index 000000000..434712687
--- /dev/null
+++ b/ci/scripts/rust_proto_check.sh
@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+# Fail if the vendored DataFusion proto files drift from the datafusion-proto /
+# datafusion-proto-common versions pinned in Cargo.lock. See the script header
of
+# dev/update_datafusion_proto.py for why these files are vendored.
+set -ex
+python3 dev/update_datafusion_proto.py --check
diff --git a/dev/update_datafusion_proto.py b/dev/update_datafusion_proto.py
new file mode 100755
index 000000000..bc627b5b8
--- /dev/null
+++ b/dev/update_datafusion_proto.py
@@ -0,0 +1,143 @@
+#!/usr/bin/env python
+
+#
+# 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.
+#
+
+# Sync the vendored DataFusion proto files with the version pinned in
Cargo.lock.
+#
+# `ballista/core/proto/ballista.proto` imports `datafusion.proto` and
+# `datafusion_common.proto`. Those two files are NOT compiled into Rust here --
+# `ballista/core/build.rs` maps their packages to the real `datafusion-proto` /
+# `datafusion-proto-common` crates via `extern_path`. They exist only so that
+# `protoc` can resolve `ballista.proto`'s imports at build time. Because of
that
+# they must stay compatible with the DataFusion version Ballista depends on.
+#
+# The `datafusion-proto` / `datafusion-proto-common` crates ship their `.proto`
+# files, but expose no `links`/`DEP_*` path for a build script to consume, so
we
+# vendor a copy. This script keeps that copy honest: it reads the exact crate
+# source resolved by `cargo metadata` and copies the `.proto` files in,
rewriting
+# the one import path that differs from our flat layout.
+#
+# Usage:
+# dev/update_datafusion_proto.py # update the vendored files in
place
+# dev/update_datafusion_proto.py --check # fail if vendored files are
stale (CI)
+
+import argparse
+import difflib
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).parent.parent.absolute()
+VENDOR_DIR = REPO_ROOT / "ballista" / "core" / "proto"
+
+# vendored file name -> (crate name, file name within the crate's `proto/` dir)
+FILES = {
+ "datafusion.proto": ("datafusion-proto", "datafusion.proto"),
+ "datafusion_common.proto": ("datafusion-proto-common",
"datafusion_common.proto"),
+}
+
+
+def crate_source_dirs():
+ """Map crate name -> source directory, as resolved in Cargo.lock."""
+ out = subprocess.run(
+ ["cargo", "metadata", "--format-version", "1", "--locked"],
+ cwd=REPO_ROOT,
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ metadata = json.loads(out.stdout)
+ dirs = {}
+ for pkg in metadata["packages"]:
+ if pkg["name"] in ("datafusion-proto", "datafusion-proto-common"):
+ # manifest_path points at the crate's Cargo.toml; protos sit
alongside it.
+ dirs[pkg["name"]] = (Path(pkg["manifest_path"]).parent,
pkg["version"])
+ missing = {"datafusion-proto", "datafusion-proto-common"} - dirs.keys()
+ if missing:
+ sys.exit(f"could not resolve crate(s) from cargo metadata:
{sorted(missing)}")
+ return dirs
+
+
+def rewrite(content: str) -> str:
+ """Rewrite crate-relative imports to our flat vendored layout.
+
+ Upstream imports the common proto by its repo path, e.g.
+ `import "datafusion/proto-common/proto/datafusion_common.proto";`
+ but we vendor it flat next to `datafusion.proto`.
+ """
+ return re.sub(
+ r'import\s+"[^"]*datafusion_common\.proto";',
+ 'import "datafusion_common.proto";',
+ content,
+ )
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ "--check",
+ action="store_true",
+ help="verify the vendored files are up to date; do not write. Exits 1
on drift.",
+ )
+ args = parser.parse_args()
+
+ dirs = crate_source_dirs()
+ stale = []
+ for vendor_name, (crate, src_name) in FILES.items():
+ crate_dir, version = dirs[crate]
+ src = crate_dir / "proto" / src_name
+ if not src.exists():
+ # Some releases don't publish the `.proto` file (e.g.
datafusion-proto-common
+ # only started shipping it in v54). Nothing to sync against, so
leave the
+ # vendored copy as-is rather than failing.
+ print(
+ f"note: {crate}@{version} does not ship proto/{src_name}; "
+ f"leaving vendored {vendor_name} unchanged",
+ file=sys.stderr,
+ )
+ continue
+ expected = rewrite(src.read_text())
+ dest = VENDOR_DIR / vendor_name
+
+ if args.check:
+ current = dest.read_text() if dest.exists() else ""
+ if current != expected:
+ stale.append(vendor_name)
+ diff = difflib.unified_diff(
+ current.splitlines(keepends=True),
+ expected.splitlines(keepends=True),
+ fromfile=f"vendored/{vendor_name}",
+ tofile=f"{crate}@{version}/proto/{src_name}",
+ )
+ sys.stderr.writelines(diff)
+ else:
+ dest.write_text(expected)
+ print(f"updated {dest.relative_to(REPO_ROOT)} from
{crate}@{version}")
+
+ if args.check and stale:
+ sys.exit(
+ "\nVendored DataFusion proto file(s) out of date: "
+ + ", ".join(stale)
+ + "\nRun `dev/update_datafusion_proto.py` and commit the result."
+ )
+
+
+if __name__ == "__main__":
+ main()
diff --git a/dev/update_datafusion_versions.py
b/dev/update_datafusion_versions.py
index aed2cee3a..d21096ea1 100755
--- a/dev/update_datafusion_versions.py
+++ b/dev/update_datafusion_versions.py
@@ -133,6 +133,14 @@ def main():
update_docs("README.md", new_version)
update_pyproject_toml(os.path.join(repo_root, "python", "pyproject.toml"),
new_version)
+ print(
+ '\nNext steps:\n'
+ ' 1. Run `cargo update` (or `cargo build`) so Cargo.lock resolves the
new '
+ 'datafusion crates.\n'
+ ' 2. Run `dev/update_datafusion_proto.py` to re-sync the vendored '
+ 'DataFusion proto files, then commit the result.'
+ )
+
if __name__ == "__main__":
main()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
