add file

cpp/src/arrow/dataset/file_csv.cc

@@ -518,7 +518,7 @@ Result<std::shared_ptr<FragmentScanOptions>> CsvFragmentScanOptions::from(
       options->parse_options.delimiter = value.data()[0];
     } else if (key == "quoting") {
       options->parse_options.quoting = parseBool(value);
-    } else if (key == "ArrowSchemaAddress") {
+    } else if (key == "column_type") {
       int64_t schema_address = std::stol(value);
       ArrowSchema* cSchema = reinterpret_cast<ArrowSchema*>(schema_address);
       ARROW_ASSIGN_OR_RAISE(auto schema, arrow::ImportSchema(cSchema));

cpp/src/arrow/engine/substrait/extension_internal.cc

@@ -0,0 +1,50 @@
+// 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.
+
+// This API is EXPERIMENTAL.
+
+#include "arrow/engine/substrait/extension_internal.h"
+
+#include "substrait/algebra.pb.h"
+
+namespace arrow {
+namespace engine {
+
+Status FromProto(const substrait::extensions::AdvancedExtension& extension,
+                 std::unordered_map<std::string, std::string>& out) {
+  if (!extension.has_enhancement()) {
+    return Status::Invalid("AdvancedExtension does not have enhancement");
+  }
+  const auto& enhancement = extension.enhancement();
+  substrait::Expression_Literal literal;
+
+  if (!enhancement.UnpackTo(&literal)) {
+    return Status::Invalid("Unpack the literal failed");
+  }
+
+  if (!literal.has_map()) {
+    return Status::Invalid("Literal does not have map");
+  }
+  auto literalMap = literal.map();
+  auto size = literalMap.key_values_size();
+  for (auto i = 0; i < size; i++) {
+    substrait::Expression_Literal_Map_KeyValue keyValue = literalMap.key_values(i);
+    out.emplace(keyValue.key().string(), keyValue.value().string());
+  }
+}
+}  // namespace engine
+}  // namespace arrow

cpp/src/arrow/engine/substrait/extension_internal.h

@@ -0,0 +1,44 @@
+
+// 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.
+
+// This API is EXPERIMENTAL.
+
+#pragma once
+
+#include <memory>
+
+#include "arrow/compute/type_fwd.h"
+#include "arrow/engine/substrait/extension_set.h"
+#include "arrow/engine/substrait/options.h"
+#include "arrow/engine/substrait/relation.h"
+#include "arrow/engine/substrait/visibility.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+
+#include "substrait/extensions/extensions.pb.h"  // IWYU pragma: export
+
+namespace arrow {
+namespace engine {
+
+/// Convert a Substrait ExtendedExpression to a vector of expressions and output names
+ARROW_ENGINE_EXPORT
+Status FromProto(const substrait::extensions::AdvancedExtension& extension,
+                 std::unordered_map<std::string, std::string>& out);
+
+}  // namespace engine
+}  // namespace arrow

java/dataset/src/main/java/org/apache/arrow/dataset/scanner/csv/CsvFragmentScanOptions.java

@@ -54,7 +54,7 @@ public ByteBuffer serialize() {
             parseOptions.entrySet().stream()),
         convertOptions.getConfigs().entrySet().stream()).collect(
         Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    options.put("ArrowSchemaAddress", Long.toString(convertOptions.getArrowSchemaAddress()));
+    options.put("column_type", Long.toString(convertOptions.getArrowSchemaAddress()));
     return serializeMap(options);
   }
   public static CsvFragmentScanOptions deserialize(String serialized) {