feat: impl stream and bulk write

.editorconfig

@@ -0,0 +1,107 @@
+# EditorConfig for GreptimeDB .NET Ingester
+# https://editorconfig.org
+
+root = true
+
+[*]
+charset = utf-8
+end_of_line = lf
+indent_style = space
+indent_size = 4
+insert_final_newline = true
+trim_trailing_whitespace = true
+
+[*.{csproj,props,targets,xml}]
+indent_size = 2
+
+[*.{json,yml,yaml}]
+indent_size = 2
+
+[*.md]
+trim_trailing_whitespace = false
+
+[*.cs]
+# Organize usings
+dotnet_sort_system_directives_first = true
+dotnet_separate_import_directive_groups = false
+
+# this. preferences
+dotnet_style_qualification_for_field = false:suggestion
+dotnet_style_qualification_for_property = false:suggestion
+dotnet_style_qualification_for_method = false:suggestion
+dotnet_style_qualification_for_event = false:suggestion
+
+# Use language keywords instead of BCL types
+dotnet_style_predefined_type_for_locals_parameters_members = true:suggestion
+dotnet_style_predefined_type_for_member_access = true:suggestion
+
+# var preferences
+csharp_style_var_for_built_in_types = false:suggestion
+csharp_style_var_when_type_is_apparent = true:suggestion
+csharp_style_var_elsewhere = true:suggestion
+
+# Expression-bodied members
+csharp_style_expression_bodied_methods = when_on_single_line:suggestion
+csharp_style_expression_bodied_constructors = false:suggestion
+csharp_style_expression_bodied_operators = when_on_single_line:suggestion
+csharp_style_expression_bodied_properties = when_on_single_line:suggestion
+csharp_style_expression_bodied_indexers = when_on_single_line:suggestion
+csharp_style_expression_bodied_accessors = when_on_single_line:suggestion
+csharp_style_expression_bodied_lambdas = when_on_single_line:suggestion
+csharp_style_expression_bodied_local_functions = when_on_single_line:suggestion
+
+# Pattern matching
+csharp_style_pattern_matching_over_is_with_cast_check = true:suggestion
+csharp_style_pattern_matching_over_as_with_null_check = true:suggestion
+
+# Null checking
+csharp_style_throw_expression = true:suggestion
+csharp_style_conditional_delegate_call = true:suggestion
+
+# Code block preferences
+csharp_prefer_braces = true:suggestion
+
+# using directive placement
+csharp_using_directive_placement = outside_namespace:warning
+
+# Namespace preferences
+csharp_style_namespace_declarations = file_scoped:warning
+
+# New line preferences
+csharp_new_line_before_open_brace = all
+csharp_new_line_before_else = true
+csharp_new_line_before_catch = true
+csharp_new_line_before_finally = true
+
+# Indentation preferences
+csharp_indent_case_contents = true
+csharp_indent_switch_labels = true
+
+# Space preferences
+csharp_space_after_cast = false
+csharp_space_after_keywords_in_control_flow_statements = true
+csharp_space_between_method_declaration_parameter_list_parentheses = false
+csharp_space_between_method_call_parameter_list_parentheses = false
+
+# Naming conventions - private instance fields should use _camelCase
+dotnet_naming_rule.private_instance_fields_should_be_camel_case.severity = warning
+dotnet_naming_rule.private_instance_fields_should_be_camel_case.symbols = private_instance_fields
+dotnet_naming_rule.private_instance_fields_should_be_camel_case.style = camel_case_underscore
+
+dotnet_naming_symbols.private_instance_fields.applicable_kinds = field
+dotnet_naming_symbols.private_instance_fields.applicable_accessibilities = private
+dotnet_naming_symbols.private_instance_fields.required_modifiers =
+
+dotnet_naming_style.camel_case_underscore.required_prefix = _
+dotnet_naming_style.camel_case_underscore.capitalization = camel_case
+
+# Static fields should use PascalCase (no underscore prefix)
+dotnet_naming_rule.static_fields_should_be_pascal_case.severity = warning
+dotnet_naming_rule.static_fields_should_be_pascal_case.symbols = static_fields
+dotnet_naming_rule.static_fields_should_be_pascal_case.style = pascal_case
+
+dotnet_naming_symbols.static_fields.applicable_kinds = field
+dotnet_naming_symbols.static_fields.applicable_accessibilities = private, internal, protected, protected_internal, private_protected
+dotnet_naming_symbols.static_fields.required_modifiers = static
+
+dotnet_naming_style.pascal_case.capitalization = pascal_case

.github/workflows/ci.yml

@@ -18,6 +18,27 @@ env:
   DOTNET_SKIP_FIRST_TIME_EXPERIENCE: true
 
 jobs:
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+        with:
+          submodules: recursive
+
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v5
+        with:
+          dotnet-version: 9.0.x
+
+      - name: Restore
+        run: dotnet restore
+
+      - name: Check Formatting
+        run: dotnet format --verify-no-changes
+
   build:
     name: Build (${{ matrix.os }})
     runs-on: ${{ matrix.os }}

Directory.Build.props

@@ -6,6 +6,7 @@
     <LangVersion>latest</LangVersion>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <AnalysisLevel>latest-recommended</AnalysisLevel>
+    <EnforceCodeStyleInBuild>true</EnforceCodeStyleInBuild>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
     <!-- CS1591: Missing XML comment, CA1720: Identifier contains type name (intentional for ColumnDataType enum) -->
     <NoWarn>$(NoWarn);CS1591;CA1720</NoWarn>

README.md

@@ -49,11 +49,52 @@ await client.DisposeAsync();
 
 ## Features
 
-- Unary write via gRPC
+- **Unary Write** - Simple single-request writes via gRPC
+- **Streaming Write** - High-throughput streaming via gRPC for multiple tables
+- **Bulk Write** - Maximum throughput via Apache Arrow Flight
 - Type coercion between .NET and GreptimeDB types
 - Health check
 - DI integration
 
+## Streaming Write
+
+For high-throughput scenarios with multiple tables:
+
+```csharp
+await using var writer = client.CreateStreamIngestWriter();
+
+// Write multiple tables in a single stream
+await writer.WriteAsync(table1);
+await writer.WriteAsync(table2);
+await writer.WriteAsync(table3);
+
+var affectedRows = await writer.CompleteAsync();
+```
+
+## Bulk Write (Arrow Flight)
+
+For maximum throughput using Apache Arrow Flight protocol:
+
+```csharp
+// Convenience helper for single-table bulk write
+var affectedRows = await client.BulkWriteAsync(table);
+```
+
+Or manage the writer lifetime yourself:
+
+```csharp
+// Note: Tables must exist before using BulkWriter
+await using var writer = client.CreateBulkWriter();
+
+await writer.WriteAsync(table);
+
+var affectedRows = await writer.CompleteAsync();
+```
+
+> **Note**: Unlike regular gRPC writes, Arrow Flight `DoPut` does not auto-create tables.
+> Ensure tables exist before using `BulkWriter`. A `BulkWriter` instance is bound to a single
+> table; create a new writer per table when bulk writing multiple tables.
+
 ## DI Integration
 
 ```csharp

src/GreptimeDB.Ingester/Arrow/ArrowTypeMapper.cs

@@ -0,0 +1,107 @@
+using Apache.Arrow;
+using Apache.Arrow.Types;
+using GreptimeDB.Ingester.Types;
+
+namespace GreptimeDB.Ingester.Arrow;
+
+/// <summary>
+/// Maps GreptimeDB column data types to Apache Arrow types.
+/// </summary>
+internal static class ArrowTypeMapper
+{
+    /// <summary>
+    /// Converts a GreptimeDB column data type to an Apache Arrow data type.
+    /// </summary>
+    /// <param name="dataType">The GreptimeDB column data type.</param>
+    /// <returns>The corresponding Apache Arrow data type.</returns>
+    /// <exception cref="NotSupportedException">Thrown when the data type is not supported.</exception>
+    public static IArrowType ToArrowType(ColumnDataType dataType)
+    {
+        return dataType switch
+        {
+            // Boolean
+            ColumnDataType.Boolean => BooleanType.Default,
+
+            // Signed integers
+            ColumnDataType.Int8 => Int8Type.Default,
+            ColumnDataType.Int16 => Int16Type.Default,
+            ColumnDataType.Int32 => Int32Type.Default,
+            ColumnDataType.Int64 => Int64Type.Default,
+
+            // Unsigned integers
+            ColumnDataType.UInt8 => UInt8Type.Default,
+            ColumnDataType.UInt16 => UInt16Type.Default,
+            ColumnDataType.UInt32 => UInt32Type.Default,
+            ColumnDataType.UInt64 => UInt64Type.Default,
+
+            // Floating point
+            ColumnDataType.Float32 => FloatType.Default,
+            ColumnDataType.Float64 => DoubleType.Default,
+
+            // String and Binary
+            ColumnDataType.String => StringType.Default,
+            ColumnDataType.Binary => BinaryType.Default,
+            ColumnDataType.Json => StringType.Default, // JSON stored as string
+
+            // Date and DateTime
+            ColumnDataType.Date => Date32Type.Default,
+            ColumnDataType.DateTime => new TimestampType(TimeUnit.Millisecond, (string?)null),
+
+            // Timestamps with different precisions
+            ColumnDataType.TimestampSecond => new TimestampType(TimeUnit.Second, (string?)null),
+            ColumnDataType.TimestampMillisecond => new TimestampType(TimeUnit.Millisecond, (string?)null),
+            ColumnDataType.TimestampMicrosecond => new TimestampType(TimeUnit.Microsecond, (string?)null),
+            ColumnDataType.TimestampNanosecond => new TimestampType(TimeUnit.Nanosecond, (string?)null),
+
+            // Time with different precisions
+            ColumnDataType.TimeSecond => new Time32Type(TimeUnit.Second),
+            ColumnDataType.TimeMillisecond => new Time32Type(TimeUnit.Millisecond),
+            ColumnDataType.TimeMicrosecond => new Time64Type(TimeUnit.Microsecond),
+            ColumnDataType.TimeNanosecond => new Time64Type(TimeUnit.Nanosecond),
+
+            _ => throw new NotSupportedException($"Unsupported data type: {dataType}")
+        };
+    }
+
+    /// <summary>
+    /// Gets the TimeUnit for a timestamp column data type.
+    /// </summary>
+    /// <param name="dataType">The GreptimeDB column data type.</param>
+    /// <returns>The corresponding TimeUnit.</returns>
+    /// <exception cref="ArgumentException">Thrown when the data type is not a timestamp type.</exception>
+    public static TimeUnit GetTimestampUnit(ColumnDataType dataType)
+    {
+        return dataType switch
+        {
+            ColumnDataType.TimestampSecond => TimeUnit.Second,
+            ColumnDataType.TimestampMillisecond => TimeUnit.Millisecond,
+            ColumnDataType.TimestampMicrosecond => TimeUnit.Microsecond,
+            ColumnDataType.TimestampNanosecond => TimeUnit.Nanosecond,
+            ColumnDataType.DateTime => TimeUnit.Millisecond,
+            _ => throw new ArgumentException($"Data type {dataType} is not a timestamp type", nameof(dataType))
+        };
+    }
+
+    /// <summary>
+    /// Checks if the data type is a timestamp type.
+    /// </summary>
+    public static bool IsTimestampType(ColumnDataType dataType)
+    {
+        return dataType is ColumnDataType.TimestampSecond
+            or ColumnDataType.TimestampMillisecond
+            or ColumnDataType.TimestampMicrosecond
+            or ColumnDataType.TimestampNanosecond
+            or ColumnDataType.DateTime;
+    }
+
+    /// <summary>
+    /// Checks if the data type is a time type.
+    /// </summary>
+    public static bool IsTimeType(ColumnDataType dataType)
+    {
+        return dataType is ColumnDataType.TimeSecond
+            or ColumnDataType.TimeMillisecond
+            or ColumnDataType.TimeMicrosecond
+            or ColumnDataType.TimeNanosecond;
+    }
+}