Merge branch 'main' into logger

Author: yingcai-cy

.github/workflows/cpp-linter.yml

@@ -39,7 +39,7 @@ jobs:
           mkdir build && cd build
           cmake .. -DCMAKE_EXPORT_COMPILE_COMMANDS=ON
           cmake --build .
-      - uses: cpp-linter/cpp-linter-action@v2.13.3
+      - uses: cpp-linter/cpp-linter-action@f91c446a32ae3eb9f98fef8c9ed4c7cb613a4f8a
         id: linter
         continue-on-error: true
         env:

cmake_modules/IcebergThirdpartyToolchain.cmake

@@ -23,9 +23,9 @@ set(ICEBERG_ARROW_INSTALL_INTERFACE_LIBS)
 # ----------------------------------------------------------------------
 # Versions and URLs for toolchain builds
 
-set(ICEBERG_ARROW_BUILD_VERSION "19.0.1")
+set(ICEBERG_ARROW_BUILD_VERSION "21.0.0")
 set(ICEBERG_ARROW_BUILD_SHA256_CHECKSUM
-    "acb76266e8b0c2fbb7eb15d542fbb462a73b3fd1e32b80fad6c2fafd95a51160")
+    "5d3f8db7e72fb9f65f4785b7a1634522e8d8e9657a445af53d4a34a3849857b5")
 
 if(DEFINED ENV{ICEBERG_ARROW_URL})
   set(ARROW_SOURCE_URL "$ENV{ICEBERG_ARROW_URL}")
@@ -65,8 +65,9 @@ function(resolve_arrow_dependency)
   set(ARROW_BUILD_STATIC
       ON
       CACHE BOOL "" FORCE)
+  # Work around undefined symbol: arrow::ipc::ReadSchema(arrow::io::InputStream*, arrow::ipc::DictionaryMemo*)
   set(ARROW_IPC
-      OFF
+      ON
       CACHE BOOL "" FORCE)
   set(ARROW_FILESYSTEM
       ON
@@ -98,10 +99,8 @@ function(resolve_arrow_dependency)
 
   fetchcontent_declare(VendoredArrow
                        ${FC_DECLARE_COMMON_OPTIONS}
-                       GIT_REPOSITORY https://github.com/apache/arrow.git
-                       GIT_TAG f12356adaaabea86638407e995e73215dbb58bb2
-                       #URL ${ARROW_SOURCE_URL}
-                       #URL_HASH "SHA256=${ICEBERG_ARROW_BUILD_SHA256_CHECKSUM}"
+                       URL ${ARROW_SOURCE_URL}
+                       URL_HASH "SHA256=${ICEBERG_ARROW_BUILD_SHA256_CHECKSUM}"
                        SOURCE_SUBDIR
                        cpp
                        FIND_PACKAGE_ARGS

src/iceberg/CMakeLists.txt

@@ -109,7 +109,11 @@ if(ICEBERG_BUILD_BUNDLE)
       avro/avro_data_util.cc
       avro/avro_reader.cc
       avro/avro_schema_util.cc
-      avro/avro_stream_internal.cc)
+      avro/avro_register.cc
+      avro/avro_stream_internal.cc
+      parquet/parquet_data_util.cc
+      parquet/parquet_reader.cc
+      parquet/parquet_schema_util.cc)
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.
   set(ICEBERG_BUNDLE_STATIC_BUILD_INTERFACE_LIBS)
@@ -161,6 +165,7 @@ if(ICEBERG_BUILD_BUNDLE)
 
   add_subdirectory(arrow)
   add_subdirectory(avro)
+  add_subdirectory(parquet)
 
   install(FILES ${CMAKE_CURRENT_BINARY_DIR}/iceberg_bundle_export.h
           DESTINATION ${ICEBERG_INSTALL_INCLUDEDIR}/iceberg)

src/iceberg/arrow/arrow_error_transform_internal.h

@@ -43,17 +43,18 @@ inline ErrorKind ToErrorKind(const ::arrow::Status& status) {
   }                                                                                   \
   lhs = std::move(result_name).ValueOrDie();
 
-#define ICEBERG_ARROW_ASSIGN_OR_RETURN(lhs, rexpr) \
-  ICEBERG_ARROW_ASSIGN_OR_RETURN_IMPL(             \
-      ARROW_ASSIGN_OR_RAISE_NAME(_error_or_value, __COUNTER__), lhs, rexpr, ToErrorKind)
-
-#define ICEBERG_ARROW_RETURN_NOT_OK(expr)                                 \
-  do {                                                                    \
-    auto&& _status = (expr);                                              \
-    if (!_status.ok()) {                                                  \
-      return std::unexpected<Error>{                                      \
-          {.kind = ToErrorKind(_status), .message = _status.ToString()}}; \
-    }                                                                     \
+#define ICEBERG_ARROW_ASSIGN_OR_RETURN(lhs, rexpr)                          \
+  ICEBERG_ARROW_ASSIGN_OR_RETURN_IMPL(                                      \
+      ARROW_ASSIGN_OR_RAISE_NAME(_error_or_value, __COUNTER__), lhs, rexpr, \
+      ::iceberg::arrow::ToErrorKind)
+
+#define ICEBERG_ARROW_RETURN_NOT_OK(expr)                                            \
+  do {                                                                               \
+    auto&& _status = (expr);                                                         \
+    if (!_status.ok()) {                                                             \
+      return std::unexpected<Error>{{.kind = ::iceberg::arrow::ToErrorKind(_status), \
+                                     .message = _status.ToString()}};                \
+    }                                                                                \
   } while (0)
 
 }  // namespace iceberg::arrow

src/iceberg/arrow/arrow_fs_file_io.cc

@@ -17,11 +17,13 @@
  * under the License.
  */
 
-#include "iceberg/arrow/arrow_fs_file_io.h"
+#include <chrono>
 
 #include <arrow/filesystem/localfs.h>
+#include <arrow/filesystem/mockfs.h>
 
 #include "iceberg/arrow/arrow_error_transform_internal.h"
+#include "iceberg/arrow/arrow_fs_file_io_internal.h"
 
 namespace iceberg::arrow {
 
@@ -67,4 +69,15 @@ Status ArrowFileSystemFileIO::DeleteFile(const std::string& file_location) {
   return {};
 }
 
+std::unique_ptr<FileIO> ArrowFileSystemFileIO::MakeMockFileIO() {
+  return std::make_unique<ArrowFileSystemFileIO>(
+      std::make_shared<::arrow::fs::internal::MockFileSystem>(
+          std::chrono::system_clock::now()));
+}
+
+std::unique_ptr<FileIO> ArrowFileSystemFileIO::MakeLocalFileIO() {
+  return std::make_unique<ArrowFileSystemFileIO>(
+      std::make_shared<::arrow::fs::LocalFileSystem>());
+}
+
 }  // namespace iceberg::arrow

src/iceberg/arrow/arrow_fs_file_io.h …ceberg/arrow/arrow_fs_file_io_internal.hsrc/iceberg/arrow/arrow_fs_file_io.h renamed to src/iceberg/arrow/arrow_fs_file_io_internal.h src/iceberg/arrow/arrow_fs_file_io.h renamed to src/iceberg/arrow/arrow_fs_file_io_internal.h

@@ -34,6 +34,12 @@ class ICEBERG_BUNDLE_EXPORT ArrowFileSystemFileIO : public FileIO {
   explicit ArrowFileSystemFileIO(std::shared_ptr<::arrow::fs::FileSystem> arrow_fs)
       : arrow_fs_(std::move(arrow_fs)) {}
 
+  /// \brief Make an in-memory FileIO backed by arrow::fs::internal::MockFileSystem.
+  static std::unique_ptr<FileIO> MakeMockFileIO();
+
+  /// \brief Make a local FileIO backed by arrow::fs::LocalFileSystem.
+  static std::unique_ptr<FileIO> MakeLocalFileIO();
+
   ~ArrowFileSystemFileIO() override = default;
 
   /// \brief Read the content of the file at the given location.

src/iceberg/avro/avro_reader.cc

@@ -31,7 +31,7 @@
 #include <avro/Generic.hh>
 #include <avro/GenericDatum.hh>
 
-#include "iceberg/arrow/arrow_fs_file_io.h"
+#include "iceberg/arrow/arrow_fs_file_io_internal.h"
 #include "iceberg/avro/avro_data_util_internal.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
 #include "iceberg/avro/avro_stream_internal.h"

src/iceberg/avro/avro_register.cc

@@ -0,0 +1,36 @@
+/*
+ * 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 "iceberg/avro/avro_register.h"
+
+#include "iceberg/avro/avro_schema_util_internal.h"
+
+namespace iceberg::avro {
+
+void RegisterLogicalTypes() {
+  static std::once_flag flag{};
+  std::call_once(flag, []() {
+    // Register the map logical type with the avro custom logical type registry.
+    // See https://github.com/apache/avro/pull/3326 for details.
+    ::avro::CustomLogicalTypeRegistry::instance().registerType(
+        "map", [](const std::string&) { return std::make_shared<MapLogicalType>(); });
+  });
+}
+
+}  // namespace iceberg::avro

src/iceberg/avro/avro_register.h

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include "iceberg/iceberg_bundle_export.h"
+
+namespace iceberg::avro {
+
+ICEBERG_BUNDLE_EXPORT void RegisterLogicalTypes();
+
+}  // namespace iceberg::avro

src/iceberg/avro/avro_schema_util.cc

@@ -22,14 +22,14 @@
 #include <string_view>
 
 #include <arrow/type.h>
-#include <arrow/util/decimal.h>
 #include <avro/CustomAttributes.hh>
 #include <avro/LogicalType.hh>
 #include <avro/NodeImpl.hh>
 #include <avro/Schema.hh>
 #include <avro/Types.hh>
 #include <avro/ValidSchema.hh>
 
+#include "iceberg/avro/avro_register.h"
 #include "iceberg/avro/avro_schema_util_internal.h"
 #include "iceberg/metadata_columns.h"
 #include "iceberg/schema.h"
@@ -49,18 +49,8 @@ constexpr std::string_view kValueIdProp = "value-id";
 constexpr std::string_view kElementIdProp = "element-id";
 constexpr std::string_view kAdjustToUtcProp = "adjust-to-utc";
 
-struct MapLogicalType : public ::avro::CustomLogicalType {
-  MapLogicalType() : ::avro::CustomLogicalType("map") {}
-};
-
 ::avro::LogicalType GetMapLogicalType() {
-  static std::once_flag flag{};
-  std::call_once(flag, []() {
-    // Register the map logical type with the avro custom logical type registry.
-    // See https://github.com/apache/avro/pull/3326 for details.
-    ::avro::CustomLogicalTypeRegistry::instance().registerType(
-        "map", [](const std::string&) { return std::make_shared<MapLogicalType>(); });
-  });
+  RegisterLogicalTypes();
   return ::avro::LogicalType(std::make_shared<MapLogicalType>());
 }