Merge pull request #11 from red-data-tools/rebane-config-section

Rename compress config_section to arrow

Author: kanga333

README.md

@@ -46,12 +46,12 @@ Example of fluent-plugin-s3 configuration.
   </format>
 
   store_as arrow
-  <compress>
+  <arrow>
     schema [
       {"name": "test_string", "type": "string"},
       {"name": "test_uint64", "type": "uint64"}
     ]
-  </compress>
+  </arrow>
 </match>
 ```
 

benchmark/prelude.rb

@@ -15,15 +15,15 @@
 ARROW_CONFIG = %[
   s3_bucket test_bucket
   store_as arrow
-  <compress>
-    arrow_format parquet
-    arrow_compression gzip
+  <arrow>
+    format parquet
+    compression gzip
     schema [
       {"name": "test_string",  "type": "string"},
       {"name": "test_uint64",  "type": "uint64"},
       {"name": "test_boolean", "type": "boolean"}
     ]
-  </compress>
+  </arrow>
 ]
 
 COLUMNIFY_CONFIG = %[

lib/fluent/plugin/s3_compressor_arrow.rb

@@ -11,29 +11,29 @@ class ArrowCompressor < Compressor
         :feather => [:gzip, :snappy],
       }
 
-      config_section :compress, multi: false do
+      config_section :arrow, multi: false do
         config_param :schema, :array
-        config_param :arrow_format, :enum, list: [:arrow, :feather, :parquet], default: :arrow
+        config_param :format, :enum, list: [:arrow, :feather, :parquet], default: :arrow
         SUPPORTED_COMPRESSION = [:gzip, :snappy, :zstd]
-        config_param :arrow_compression, :enum, list: SUPPORTED_COMPRESSION, default: nil
-        config_param :arrow_chunk_size, :integer, default: 1024
+        config_param :compression, :enum, list: SUPPORTED_COMPRESSION, default: nil
+        config_param :chunk_size, :integer, default: 1024
       end
 
       def configure(conf)
         super
 
-        if INVALID＿COMBINATIONS[@compress.arrow_format]&.include? @compress.arrow_compression
-          raise Fluent::ConfigError, "#{@compress.arrow_format} unsupported with #{@compress.arrow_format}"
+        if INVALID＿COMBINATIONS[@arrow.format]&.include? @arrow.compression
+          raise Fluent::ConfigError, "#{@arrow.format} unsupported with #{@arrow.format}"
         end
 
-        @arrow_schema = Arrow::Schema.new(@compress.schema)
+        @schema = Arrow::Schema.new(@arrow.schema)
         @options = Arrow::JSONReadOptions.new
-        @options.schema = @arrow_schema
+        @options.schema = @schema
         @options.unexpected_field_behavior = :ignore
       end
 
       def ext
-        @compress.arrow_format.freeze
+        @arrow.format.freeze
       end
 
       def content_type
@@ -46,9 +46,9 @@ def compress(chunk, tmp)
         table = Arrow::JSONReader.new(stream, @options)
 
         table.read.save(tmp,
-          format: @compress.arrow_format,
-          chunk_size: @compress.arrow_chunk_size,
-          compression: @compress.arrow_compression,
+          format: @arrow.format,
+          chunk_size: @arrow.chunk_size,
+          compression: @arrow.compression,
         )
       end
     end

test/plugin/test_s3_compressor_arrow.rb

@@ -14,15 +14,15 @@ def setup
       {"name": "test_string", "type": "string"},
       {"name": "test_uint64", "type": "uint64"},
     ]
-    CONFIG = config_element("ROOT", "", S3_CONFIG, [config_element("compress", "", {"schema" => SCHEMA})])
+    CONFIG = config_element("ROOT", "", S3_CONFIG, [config_element("arrow", "", {"schema" => SCHEMA})])
 
     def test_configure
       d = create_driver
       c = d.instance.instance_variable_get(:@compressor)
       assert_equal :arrow, c.ext
       assert_equal 'application/x-apache-arrow-file', c.content_type
-      assert c.instance_variable_get(:@arrow_schema).is_a?(Arrow::Schema)
-      assert_equal 1024, c.instance_variable_get(:@compress).arrow_chunk_size
+      assert c.instance_variable_get(:@schema).is_a?(Arrow::Schema)
+      assert_equal 1024, c.instance_variable_get(:@arrow).chunk_size
     end
 
     data(
@@ -32,9 +32,9 @@ def test_configure
     )
     def test_invalid_configure
       format, compression = data
-      arrow_config = config_element("compress", "", { "schema" => SCHEMA,
-        "arrow_format" => format,
-        "arrow_compression" => compression,
+      arrow_config = config_element("arrow", "", { "schema" => SCHEMA,
+        "format" => format,
+        "compression" => compression,
       })
       config = config_element("ROOT", "", S3_CONFIG, [arrow_config])
       assert_raise Fluent::ConfigError do
@@ -65,9 +65,9 @@ def test_compress
     end
 
     data(gzip: "gzip", zstd: "zstd")
-    def test_compress_with_arrow_compression
-      arrow_config = config_element("compress", "", { "schema" => SCHEMA,
-        "arrow_compression" => data,
+    def test_compress_with_compression
+      arrow_config = config_element("arrow", "", { "schema" => SCHEMA,
+        "compression" => data,
       })
       config = config_element("ROOT", "", S3_CONFIG, [arrow_config])
 
@@ -100,9 +100,9 @@ def test_compress_with_arrow_compression
     )
     def test_compress_with_format
       format, compression = data
-      arrow_config = config_element("compress", "", { "schema" => SCHEMA,
-        "arrow_format" => format,
-        "arrow_compression" => compression,
+      arrow_config = config_element("arrow", "", { "schema" => SCHEMA,
+        "format" => format,
+        "compression" => compression,
       })
       config = config_element("ROOT", "", S3_CONFIG, [arrow_config])