Add StorageIceberg and table function iceberg

src/IO/S3Common.cpp

@@ -17,8 +17,11 @@
 #    include <aws/core/auth/AWSCredentialsProviderChain.h>
 #    include <aws/core/auth/STSCredentialsProvider.h>
 #    include <aws/core/client/SpecifiedRetryableErrorsRetryStrategy.h>
+#    include <aws/core/http/HttpClientFactory.h>
 #    include <aws/core/platform/Environment.h>
 #    include <aws/core/platform/OSVersionInfo.h>
+#    include <aws/core/utils/HashingUtils.h>
+#    include <aws/core/utils/UUID.h>
 #    include <aws/core/utils/json/JsonSerializer.h>
 #    include <aws/core/utils/logging/LogMacros.h>
 #    include <aws/core/utils/logging/LogSystemInterface.h>
@@ -798,8 +801,49 @@ namespace S3
             get_request_throttler,
             put_request_throttler);
     }
-}
 
+    std::vector<String>
+    listFiles(const Aws::S3::S3Client & client, const String & bucket, const String & key, const String & prefix, const String & extension)
+    {
+        std::vector<String> res;
+        S3::ListObjectsV2Request request;
+        Aws::S3::Model::ListObjectsV2Outcome outcome;
+
+        bool is_finished{false};
+
+        request.SetBucket(bucket);
+
+        request.SetPrefix(prefix);
+
+        while (!is_finished)
+        {
+            outcome = client.ListObjectsV2(request);
+            if (!outcome.IsSuccess())
+                throw Exception(
+                    ErrorCodes::S3_ERROR,
+                    "Could not list objects in bucket {} with key {}, S3 exception: {}, message: {}",
+                    quoteString(bucket),
+                    quoteString(key),
+                    backQuote(outcome.GetError().GetExceptionName()),
+                    quoteString(outcome.GetError().GetMessage()));
+
+            const auto & result_batch = outcome.GetResult().GetContents();
+            for (const auto & obj : result_batch)
+            {
+                const auto & filename = obj.GetKey();
+
+                if (std::filesystem::path(filename).extension() == extension)
+                    res.push_back(filename);
+            }
+
+            request.SetContinuationToken(outcome.GetResult().GetNextContinuationToken());
+
+            is_finished = !outcome.GetResult().GetIsTruncated();
+        }
+
+        return res;
+    }
+}
 }
 
 #endif

src/IO/S3Common.h

@@ -97,6 +97,8 @@ class ClientFactory
     std::atomic<bool> s3_requests_logging_enabled;
 };
 
+std::vector<String>
+listFiles(const Aws::S3::S3Client & client, const String & bucket, const String & key, const String & prefix, const String & extension);
 }
 #endif
 

src/Processors/Formats/Impl/AvroRowInputFormat.cpp

@@ -68,35 +68,35 @@ namespace ErrorCodes
     extern const int CANNOT_READ_ALL_DATA;
 }
 
-class InputStreamReadBufferAdapter : public avro::InputStream
+bool AvroInputStreamReadBufferAdapter::next(const uint8_t ** data, size_t * len)
 {
-public:
-    explicit InputStreamReadBufferAdapter(ReadBuffer & in_) : in(in_) {}
-
-    bool next(const uint8_t ** data, size_t * len) override
+    if (in.eof())
     {
-        if (in.eof())
-        {
-            *len = 0;
-            return false;
-        }
-
-        *data = reinterpret_cast<const uint8_t *>(in.position());
-        *len = in.available();
-
-        in.position() += in.available();
-        return true;
+        *len = 0;
+        return false;
     }
 
-    void backup(size_t len) override { in.position() -= len; }
+    *data = reinterpret_cast<const uint8_t *>(in.position());
+    *len = in.available();
 
-    void skip(size_t len) override { in.tryIgnore(len); }
+    in.position() += in.available();
+    return true;
+}
 
-    size_t byteCount() const override { return in.count(); }
+void AvroInputStreamReadBufferAdapter::backup(size_t len)
+{
+    in.position() -= len;
+}
 
-private:
-    ReadBuffer & in;
-};
+void AvroInputStreamReadBufferAdapter::skip(size_t len)
+{
+    in.tryIgnore(len);
+}
+
+size_t AvroInputStreamReadBufferAdapter::byteCount() const
+{
+    return in.count();
+}
 
 /// Insert value with conversion to the column of target type.
 template <typename T>
@@ -757,7 +757,7 @@ AvroRowInputFormat::AvroRowInputFormat(const Block & header_, ReadBuffer & in_,
 
 void AvroRowInputFormat::readPrefix()
 {
-    file_reader_ptr = std::make_unique<avro::DataFileReaderBase>(std::make_unique<InputStreamReadBufferAdapter>(*in));
+    file_reader_ptr = std::make_unique<avro::DataFileReaderBase>(std::make_unique<AvroInputStreamReadBufferAdapter>(*in));
     deserializer_ptr = std::make_unique<AvroDeserializer>(
         output.getHeader(), file_reader_ptr->dataSchema(), format_settings.avro.allow_missing_fields, format_settings.avro.null_as_default);
     file_reader_ptr->init();
@@ -914,7 +914,7 @@ AvroConfluentRowInputFormat::AvroConfluentRowInputFormat(
 
 void AvroConfluentRowInputFormat::readPrefix()
 {
-    input_stream = std::make_unique<InputStreamReadBufferAdapter>(*in);
+    input_stream = std::make_unique<AvroInputStreamReadBufferAdapter>(*in);
     decoder = avro::binaryDecoder();
     decoder->init(*input_stream);
 }
@@ -971,7 +971,7 @@ NamesAndTypesList AvroSchemaReader::readSchema()
     }
     else
     {
-        auto file_reader_ptr = std::make_unique<avro::DataFileReaderBase>(std::make_unique<InputStreamReadBufferAdapter>(in));
+        auto file_reader_ptr = std::make_unique<avro::DataFileReaderBase>(std::make_unique<AvroInputStreamReadBufferAdapter>(in));
         root_node = file_reader_ptr->dataSchema().root();
     }
 

src/Processors/Formats/Impl/AvroRowInputFormat.h

@@ -28,6 +28,23 @@ namespace ErrorCodes
     extern const int INCORRECT_DATA;
 }
 
+class AvroInputStreamReadBufferAdapter : public avro::InputStream
+{
+public:
+    explicit AvroInputStreamReadBufferAdapter(ReadBuffer & in_) : in(in_) {}
+
+    bool next(const uint8_t ** data, size_t * len) override;
+
+    void backup(size_t len) override;
+
+    void skip(size_t len) override;
+
+    size_t byteCount() const override;
+
+private:
+    ReadBuffer & in;
+};
+
 class AvroDeserializer
 {
 public:
@@ -185,8 +202,8 @@ class AvroSchemaReader : public ISchemaReader
 
     NamesAndTypesList readSchema() override;
 
+    static DataTypePtr avroNodeToDataType(avro::NodePtr node);
 private:
-    DataTypePtr avroNodeToDataType(avro::NodePtr node);
 
     bool confluent;
     const FormatSettings format_settings;