Merge pull request #10 from red-data-tools/fix-not-using-schema

Fix not using schema

Author: kanga333

lib/fluent/plugin/s3_compressor_arrow.rb

@@ -22,10 +22,14 @@ class ArrowCompressor < Compressor
       def configure(conf)
         super
 
-        @arrow_schema = ::Arrow::Schema.new(@compress.schema)
         if INVALID＿COMBINATIONS[@compress.arrow_format]&.include? @compress.arrow_compression
           raise Fluent::ConfigError, "#{@compress.arrow_format} unsupported with #{@compress.arrow_format}"
         end
+
+        @arrow_schema = Arrow::Schema.new(@compress.schema)
+        @options = Arrow::JSONReadOptions.new
+        @options.schema = @arrow_schema
+        @options.unexpected_field_behavior = :ignore
       end
 
       def ext
@@ -39,7 +43,7 @@ def content_type
       def compress(chunk, tmp)
         buffer = Arrow::Buffer.new(chunk.read)
         stream = Arrow::BufferInputStream.new(buffer)
-        table = Arrow::JSONReader.new(stream)
+        table = Arrow::JSONReader.new(stream, @options)
 
         table.read.save(tmp,
           format: @compress.arrow_format,

test/plugin/test_s3_compressor_arrow.rb

@@ -48,15 +48,17 @@ def test_compress
 
       chunk = Fluent::Plugin::Buffer::MemoryChunk.new(Object.new)
       d1 = {"test_string" => 'record1', "test_uint64" => 1}
-      d2 = {"test_string" => 'record2', "test_uint64" => 2}
+      d2 = {"test_string" => 'record2', "test_uint64" => 2, "unexpected_field" => false}
+      expected_d2 = d2.dup
+      expected_d2.delete "unexpected_field"
       chunk.append([d1.to_json + "\n", d2.to_json + "\n"])
 
       Tempfile.create do |tmp|
         c.compress(chunk, tmp)
         Arrow::MemoryMappedInputStream.open(tmp.path) do |input|
           reader = Arrow::RecordBatchFileReader.new(input)
           reader.each do |record_batch|
-            assert_equal([d1, d2], record_batch.collect(&:to_h))
+            assert_equal([d1, expected_d2], record_batch.collect(&:to_h))
           end
         end
       end