[VL] Validate parquet field ids on the Iceberg write path

backends-velox/src/main/resources/org/apache/gluten/proto/ParquetFieldIds.proto

@@ -0,0 +1,48 @@
+// SPDX-License-Identifier: Apache-2.0
+syntax = "proto3";
+
+package gluten;
+
+option java_package = "org.apache.gluten.proto";
+option java_multiple_files = true;
+
+// An explicit Parquet field-id annotation for one node of a write schema.
+//
+// The tree is shaped as the LOGICAL (Spark/Velox) type tree, not the Parquet
+// physical tree:
+//   struct    -> one child per field
+//   array     -> exactly one child, the synthetic element node
+//   map       -> exactly two children, the synthetic key node then the value node
+//   primitive -> no children
+// This is what velox::parquet::Writer expects; see
+// velox/dwio/parquet/writer/Writer.cpp validateSchemaRecursive.
+//
+// `name` lets the native side JOIN the tree against the Velox RowType instead of
+// zipping it by index, so a producer that derives the tree separately from the
+// schema (the Iceberg write path does) cannot silently mis-assign ids by
+// reordering. It is empty on synthetic array-element / map-key / map-value nodes,
+// which have no name in the Spark type tree.
+//
+// `id` uses proto3 field presence: an ABSENT id means "this node legitimately has
+// no field id" (for example a Delta row-tracking column), which is a different
+// statement from any sentinel value. Do not reuse IcebergNestedField for this --
+// its `int32 id = 1` has no presence and therefore cannot express absence.
+message ParquetFieldIdNode {
+  enum Kind {
+    PRIMITIVE = 0;
+    STRUCT = 1;
+    ARRAY = 2;
+    MAP = 3;
+  }
+
+  string name = 1;
+  Kind kind = 2;
+  optional int32 id = 3;
+  repeated ParquetFieldIdNode children = 4;
+}
+
+// The field-id annotation for a whole write schema: one node per top-level column,
+// in write-schema order.
+message ParquetFieldIdSchema {
+  repeated ParquetFieldIdNode fields = 1;
+}

cpp/velox/CMakeLists.txt

@@ -212,6 +212,7 @@ set(VELOX_SRCS
     utils/VeloxArrowUtils.cc
     utils/VeloxBatchResizer.cc
     utils/VeloxWholeStageDumper.cc
+    utils/ParquetFieldIds.cc
     utils/VeloxWriterUtils.cc)
 
 if(ENABLE_S3)

cpp/velox/compute/iceberg/IcebergWriter.cc

@@ -24,6 +24,7 @@
 #include "compute/iceberg/IcebergFormat.h"
 #include "config/VeloxConfig.h"
 #include "utils/ConfigExtractor.h"
+#include "utils/ParquetFieldIds.h"
 #include "velox/connectors/hive/iceberg/IcebergDataSink.h"
 #include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
 
@@ -102,19 +103,6 @@ class GlutenIcebergFileNameGenerator : public connector::hive::FileNameGenerator
   mutable int32_t fileCount_;
 };
 
-parquet::ParquetFieldId convertToIcebergNestedField(const gluten::IcebergNestedField& protoField) {
-  parquet::ParquetFieldId result;
-  result.fieldId = protoField.id();
-
-  // Recursively convert children
-  result.children.reserve(protoField.children_size());
-  for (const auto& protoChild : protoField.children()) {
-    result.children.push_back(convertToIcebergNestedField(protoChild));
-  }
-
-  return result;
-}
-
 std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
     const RowTypePtr& outputRowType,
     const std::string& outputDirectoryPath,
@@ -124,7 +112,7 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
     int64_t taskId,
     const std::string& operationId,
     std::shared_ptr<const IcebergPartitionSpec> spec,
-    const parquet::ParquetFieldId& nestedField,
+    const std::vector<parquet::ParquetFieldId>& fieldIds,
     facebook::velox::memory::MemoryPool* pool) {
   std::vector<std::shared_ptr<const iceberg::IcebergColumnHandle>> columnHandles;
 
@@ -142,13 +130,13 @@ std::shared_ptr<IcebergInsertTableHandle> createIcebergInsertTableHandle(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kPartitionKey,
           columnTypes.at(i),
-          nestedField.children[i]));
+          fieldIds.at(i)));
     } else {
       columnHandles.push_back(std::make_shared<iceberg::IcebergColumnHandle>(
           columnNames.at(i),
           connector::hive::HiveColumnHandle::ColumnType::kRegular,
           columnTypes.at(i),
-          nestedField.children[i]));
+          fieldIds.at(i)));
     }
   }
 
@@ -189,7 +177,7 @@ IcebergWriter::IcebergWriter(
     std::shared_ptr<facebook::velox::memory::MemoryPool> memoryPool,
     std::shared_ptr<facebook::velox::memory::MemoryPool> connectorPool)
     : rowType_(rowType),
-      field_(convertToIcebergNestedField(field)),
+      fieldIds_(gluten::resolveParquetFieldIds(gluten::fromProto(field), rowType, /*checkNames=*/false)),
       partitionId_(partitionId),
       taskId_(taskId),
       operationId_(operationId),
@@ -242,7 +230,7 @@ IcebergWriter::IcebergWriter(
           taskId_,
           operationId_,
           spec,
-          field_,
+          fieldIds_,
           pool_.get()),
       connectorQueryCtx_.get(),
       facebook::velox::connector::CommitStrategy::kNoCommit,

cpp/velox/compute/iceberg/IcebergWriter.h

@@ -20,6 +20,7 @@
 #include "IcebergNestedField.pb.h"
 #include "memory/VeloxColumnarBatch.h"
 #include "utils/Metrics.h"
+#include "utils/ParquetFieldIds.h"
 #include "velox/connectors/hive/iceberg/IcebergColumnHandle.h"
 #include "velox/connectors/hive/iceberg/IcebergDataSink.h"
 
@@ -60,7 +61,9 @@ class IcebergWriter {
 
  private:
   facebook::velox::RowTypePtr rowType_;
-  const facebook::velox::parquet::ParquetFieldId field_;
+  // Resolved against rowType_ and validated (arity, positivity, uniqueness) at
+  // construction, so it is safe to index by column ordinal.
+  const std::vector<facebook::velox::parquet::ParquetFieldId> fieldIds_;
   int32_t partitionId_;
   int64_t taskId_;
   std::string operationId_;

cpp/velox/tests/CMakeLists.txt

@@ -145,6 +145,7 @@ if(ENABLE_S3)
 endif()
 add_velox_test(scoped_timer_test SOURCES ScopedTimerTest.cc)
 add_velox_test(row_based_checksum_test SOURCES RowBasedChecksumTest.cc)
+add_velox_test(parquet_field_ids_test SOURCES utils/ParquetFieldIdsTest.cc)
 if(BUILD_EXAMPLES)
   add_velox_test(my_udf_test SOURCES MyUdfTest.cc)
 endif()

cpp/velox/tests/utils/ParquetFieldIdsTest.cc

@@ -0,0 +1,242 @@
+/*
+ * 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.
+ */
+
+#include "utils/ParquetFieldIds.h"
+
+#include <gtest/gtest.h>
+
+#include "ParquetFieldIds.pb.h"
+#include "utils/Exception.h"
+#include "velox/type/Type.h"
+
+using namespace facebook::velox;
+using namespace gluten;
+
+namespace {
+
+// Builders keep the tests readable; every one of them mirrors what the JVM walker
+// is specified to emit.
+FieldIdNode prim(const std::string& name, std::optional<int32_t> id) {
+  return FieldIdNode{name, FieldIdNode::Kind::kPrimitive, id, {}};
+}
+
+FieldIdNode strct(const std::string& name, std::optional<int32_t> id, std::vector<FieldIdNode> children) {
+  return FieldIdNode{name, FieldIdNode::Kind::kStruct, id, std::move(children)};
+}
+
+FieldIdNode arr(const std::string& name, std::optional<int32_t> id, FieldIdNode element) {
+  return FieldIdNode{name, FieldIdNode::Kind::kArray, id, {std::move(element)}};
+}
+
+FieldIdNode map(const std::string& name, std::optional<int32_t> id, FieldIdNode key, FieldIdNode value) {
+  return FieldIdNode{name, FieldIdNode::Kind::kMap, id, {std::move(key), std::move(value)}};
+}
+
+std::string resolveError(const std::vector<FieldIdNode>& nodes, const RowTypePtr& rowType, bool checkNames = true) {
+  try {
+    resolveParquetFieldIds(nodes, rowType, checkNames);
+  } catch (const GlutenException& e) {
+    return e.what();
+  }
+  return "";
+}
+
+} // namespace
+
+TEST(ParquetFieldIdsTest, flatSchema) {
+  auto rowType = ROW({"a", "b"}, {INTEGER(), VARCHAR()});
+  auto ids = resolveParquetFieldIds({prim("a", 1), prim("b", 2)}, rowType, true);
+  ASSERT_EQ(ids.size(), 2);
+  EXPECT_EQ(ids[0].fieldId, 1);
+  EXPECT_EQ(ids[1].fieldId, 2);
+  EXPECT_TRUE(ids[0].children.empty());
+}
+
+TEST(ParquetFieldIdsTest, nestedStruct) {
+  auto rowType = ROW({"s"}, {ROW({"x", "y"}, {INTEGER(), INTEGER()})});
+  auto ids = resolveParquetFieldIds({strct("s", 1, {prim("x", 2), prim("y", 3)})}, rowType, true);
+  ASSERT_EQ(ids.size(), 1);
+  EXPECT_EQ(ids[0].fieldId, 1);
+  ASSERT_EQ(ids[0].children.size(), 2);
+  EXPECT_EQ(ids[0].children[0].fieldId, 2);
+  EXPECT_EQ(ids[0].children[1].fieldId, 3);
+}
+
+TEST(ParquetFieldIdsTest, arrayHasExactlyOneChild) {
+  auto rowType = ROW({"a"}, {ARRAY(INTEGER())});
+  auto ids = resolveParquetFieldIds({arr("a", 1, prim("", 2))}, rowType, true);
+  ASSERT_EQ(ids.size(), 1);
+  ASSERT_EQ(ids[0].children.size(), 1);
+  EXPECT_EQ(ids[0].children[0].fieldId, 2);
+}
+
+TEST(ParquetFieldIdsTest, mapHasKeyThenValue) {
+  auto rowType = ROW({"m"}, {MAP(VARCHAR(), INTEGER())});
+  auto ids = resolveParquetFieldIds({map("m", 1, prim("", 2), prim("", 3))}, rowType, true);
+  ASSERT_EQ(ids[0].children.size(), 2);
+  EXPECT_EQ(ids[0].children[0].fieldId, 2);
+  EXPECT_EQ(ids[0].children[1].fieldId, 3);
+}
+
+TEST(ParquetFieldIdsTest, arrayOfStruct) {
+  auto rowType = ROW({"items"}, {ARRAY(ROW({"sku", "qty"}, {VARCHAR(), INTEGER()}))});
+  auto ids = resolveParquetFieldIds({arr("items", 5, strct("", 6, {prim("sku", 7), prim("qty", 8)}))}, rowType, true);
+  ASSERT_EQ(ids[0].children.size(), 1);
+  ASSERT_EQ(ids[0].children[0].children.size(), 2);
+  EXPECT_EQ(ids[0].children[0].children[1].fieldId, 8);
+}
+
+// The two cases velox's own validateSchemaRecursive never reaches: it recurses into
+// an array element only when that element is a ROW, so a short tree under a nested
+// array or a map reaches an unguarded .at(0) instead of a diagnostic.
+TEST(ParquetFieldIdsTest, nestedArrayShortTreeIsRejected) {
+  auto rowType = ROW({"a"}, {ARRAY(ARRAY(INTEGER()))});
+  auto bad = arr("a", 1, FieldIdNode{"", FieldIdNode::Kind::kArray, 2, {}}); // inner array missing its element
+  auto err = resolveError({bad}, rowType);
+  EXPECT_NE(err.find("a.element"), std::string::npos) << err;
+  EXPECT_NE(err.find("exactly one child"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, arrayOfMapShortTreeIsRejected) {
+  auto rowType = ROW({"a"}, {ARRAY(MAP(VARCHAR(), INTEGER()))});
+  auto bad = arr("a", 1, FieldIdNode{"", FieldIdNode::Kind::kMap, 2, {prim("", 3)}}); // map with one child
+  auto err = resolveError({bad}, rowType);
+  EXPECT_NE(err.find("a.element"), std::string::npos) << err;
+  EXPECT_NE(err.find("exactly two children"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, nestedArrayHappyPath) {
+  auto rowType = ROW({"a"}, {ARRAY(ARRAY(INTEGER()))});
+  auto ids = resolveParquetFieldIds({arr("a", 1, arr("", 2, prim("", 3)))}, rowType, true);
+  EXPECT_EQ(ids[0].children[0].children[0].fieldId, 3);
+}
+
+TEST(ParquetFieldIdsTest, tooFewTopLevelNodesIsRejected) {
+  auto rowType = ROW({"a", "b"}, {INTEGER(), INTEGER()});
+  auto err = resolveError({prim("a", 1)}, rowType);
+  EXPECT_NE(err.find("one child per struct field"), std::string::npos) << err;
+}
+
+// velox guards with '<=', so it silently accepts and ignores extra children.
+TEST(ParquetFieldIdsTest, tooManyNodesIsRejected) {
+  auto rowType = ROW({"a"}, {INTEGER()});
+  auto err = resolveError({prim("a", 1), prim("b", 2)}, rowType);
+  EXPECT_NE(err.find("got 2"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, nameMismatchIsRejectedWhenCheckingNames) {
+  auto rowType = ROW({"a", "b"}, {INTEGER(), INTEGER()});
+  auto err = resolveError({prim("a", 1), prim("wrong", 2)}, rowType);
+  EXPECT_NE(err.find("'wrong'"), std::string::npos) << err;
+  EXPECT_NE(err.find("'b'"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, nameMismatchIsToleratedWhenNotCheckingNames) {
+  // The Iceberg wire format carries no names.
+  auto rowType = ROW({"a", "b"}, {INTEGER(), INTEGER()});
+  auto ids = resolveParquetFieldIds({prim("", 1), prim("", 2)}, rowType, false);
+  EXPECT_EQ(ids[1].fieldId, 2);
+}
+
+TEST(ParquetFieldIdsTest, kindMismatchIsRejected) {
+  auto rowType = ROW({"a"}, {ARRAY(INTEGER())});
+  auto err = resolveError({prim("a", 1)}, rowType);
+  EXPECT_NE(err.find("PRIMITIVE"), std::string::npos) << err;
+  EXPECT_NE(err.find("ARRAY"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, absentIdLowersToSentinelNotZero) {
+  // A Delta row-tracking column legitimately has no field id. It must reach velox
+  // as a strictly negative value: fieldIdMetadata() emits metadata for id >= 0, so
+  // 0 would be written to the footer as a real field id.
+  auto rowType = ROW({"a", "internal"}, {INTEGER(), INTEGER()});
+  auto ids = resolveParquetFieldIds({prim("a", 1), prim("internal", std::nullopt)}, rowType, true);
+  EXPECT_EQ(ids[0].fieldId, 1);
+  EXPECT_EQ(ids[1].fieldId, kNoParquetFieldId);
+  EXPECT_LT(ids[1].fieldId, 0);
+}
+
+TEST(ParquetFieldIdsTest, zeroIdIsRejected) {
+  auto rowType = ROW({"a"}, {INTEGER()});
+  auto err = resolveError({prim("a", 0)}, rowType);
+  EXPECT_NE(err.find("positive"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, negativeIdIsRejected) {
+  auto rowType = ROW({"a"}, {INTEGER()});
+  auto err = resolveError({prim("a", -1)}, rowType);
+  EXPECT_NE(err.find("positive"), std::string::npos) << err;
+}
+
+// Global, not per-level. velox checks neither.
+TEST(ParquetFieldIdsTest, duplicateIdAcrossSiblingStructsIsRejected) {
+  auto rowType = ROW({"s1", "s2"}, {ROW({"x"}, {INTEGER()}), ROW({"y"}, {INTEGER()})});
+  auto err = resolveError({strct("s1", 1, {prim("x", 5)}), strct("s2", 2, {prim("y", 5)})}, rowType);
+  EXPECT_NE(err.find("Duplicate"), std::string::npos) << err;
+  EXPECT_NE(err.find("s1.x"), std::string::npos) << err;
+  EXPECT_NE(err.find("s2.y"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, errorNamesTheDottedPath) {
+  auto rowType = ROW({"payload"}, {ROW({"items"}, {ARRAY(ROW({"sku"}, {VARCHAR()}))})});
+  auto bad = strct("payload", 1, {arr("items", 2, strct("", 3, {}))}); // struct element missing 'sku'
+  auto err = resolveError({bad}, rowType);
+  EXPECT_NE(err.find("payload.items.element"), std::string::npos) << err;
+}
+
+TEST(ParquetFieldIdsTest, emptySchemaResolvesToEmptyVector) {
+  auto rowType = ROW({}, {});
+  auto ids = resolveParquetFieldIds({}, rowType, true);
+  EXPECT_TRUE(ids.empty());
+}
+
+TEST(ParquetFieldIdsTest, decodesFromProtoWithFieldPresence) {
+  ParquetFieldIdSchema schema;
+  auto* a = schema.add_fields();
+  a->set_name("a");
+  a->set_kind(ParquetFieldIdNode::PRIMITIVE);
+  a->set_id(7);
+  auto* b = schema.add_fields();
+  b->set_name("b");
+  b->set_kind(ParquetFieldIdNode::PRIMITIVE);
+  // no id set -> absence, which must survive decoding as absence
+
+  auto nodes = fromProto(schema);
+  ASSERT_EQ(nodes.size(), 2);
+  ASSERT_TRUE(nodes[0].id.has_value());
+  EXPECT_EQ(nodes[0].id.value(), 7);
+  EXPECT_FALSE(nodes[1].id.has_value());
+}
+
+TEST(ParquetFieldIdsTest, decodesNestedProto) {
+  ParquetFieldIdSchema schema;
+  auto* items = schema.add_fields();
+  items->set_name("items");
+  items->set_kind(ParquetFieldIdNode::ARRAY);
+  items->set_id(1);
+  auto* element = items->add_children();
+  element->set_kind(ParquetFieldIdNode::STRUCT);
+  element->set_id(2);
+  auto* sku = element->add_children();
+  sku->set_name("sku");
+  sku->set_kind(ParquetFieldIdNode::PRIMITIVE);
+  sku->set_id(3);
+
+  auto rowType = ROW({"items"}, {ARRAY(ROW({"sku"}, {VARCHAR()}))});
+  auto ids = resolveParquetFieldIds(fromProto(schema), rowType, true);
+  EXPECT_EQ(ids[0].children[0].children[0].fieldId, 3);
+}