You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "eric-wang-1990 (via GitHub)" <gi...@apache.org> on 2025/04/03 08:32:37 UTC
[PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
eric-wang-1990 opened a new pull request, #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669
1. Create a new file [Lz4Utilities.cs](https://github.com/apache/arrow-adbc/compare/main...eric-wang-1990:arrow-adbc:add_lz4_to_arrowbatch#diff-7275463b6b9fcc3cf3b954580e3ded8b0b7237a90e0f4aea33eb11e613f3db39) to abstract common Lz4 decompress util functions for both cloud fetch and arrow batch.
2. Add support for decompress arrow batch with Lz4
3. Rename adbc.spark.cloudfetch.lz4.enabled to adbc.spark.lz4Compression.enabled since it is not specific to cloudfetch
4. Add test to test both cloudFetch and arrowBatch in StatementTests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "eric-wang-1990 (via GitHub)" <gi...@apache.org>.
eric-wang-1990 commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027721807
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
+ dataToUse = dataStream.ToArray();
Review Comment:
Sounds good, will change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "eric-wang-1990 (via GitHub)" <gi...@apache.org>.
eric-wang-1990 commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027735213
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
+ dataToUse = dataStream.ToArray();
+ dataStream.Dispose();
+ }
+ catch (Exception ex)
+ {
+ // If decompression fails, use the original data
+ System.Diagnostics.Debug.WriteLine($"Failed to decompress LZ4 data: {ex.Message}");
+ }
+ }
+
+ // Always use ChunkStream which ensures proper schema handling
+ this.reader = new ArrowStreamReader(new ChunkStream(this.schema, dataToUse));
+ }
+ catch (Exception ex)
+ {
+ // Log any errors and skip this batch
+ System.Diagnostics.Debug.WriteLine($"Error processing batch: {ex.Message}");
Review Comment:
Yeah I was planning to throw, somehow this part got ommited. We do not want partial data and should definitely throw here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "eric-wang-1990 (via GitHub)" <gi...@apache.org>.
eric-wang-1990 commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027731089
##########
csharp/src/Drivers/Apache/Spark/Lz4Utilities.cs:
##########
@@ -0,0 +1,57 @@
+/*
+* 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.
+*/
+
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using K4os.Compression.LZ4.Streams;
+
+namespace Apache.Arrow.Adbc.Drivers.Apache.Spark
+{
+ /// <summary>
+ /// Utility class for LZ4 compression/decompression operations.
+ /// </summary>
+ internal static class Lz4Utilities
+ {
+ /// <summary>
+ /// Decompresses LZ4 compressed data into a memory stream.
+ /// </summary>
+ /// <param name="compressedData">The compressed data bytes.</param>
+ /// <param name="cancellationToken">A cancellation token.</param>
+ /// <returns>A memory stream containing the decompressed data, positioned at the beginning of the stream.</returns>
+ /// <exception cref="AdbcException">Thrown when decompression fails.</exception>
+ public static async Task<MemoryStream> DecompressLz4Async(byte[] compressedData, CancellationToken cancellationToken = default)
Review Comment:
Makes sense, will change to sync.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "eric-wang-1990 (via GitHub)" <gi...@apache.org>.
eric-wang-1990 commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027736544
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
Review Comment:
This is not needed since now it will return a ReadOnlyMemory
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher merged PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#issuecomment-2775945220
> Just curious, does Spark/Databricks not use Arrow IPC compression? It compresses the file as a whole separately?
Yeah I asked a similar question in a previous PR, but apparently it doesn't (or it wouldn't need a separate out-of-band flag for the compression setting).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "lidavidm (via GitHub)" <gi...@apache.org>.
lidavidm commented on PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#issuecomment-2775281114
Just curious, does Spark/Databricks not use Arrow IPC compression? It compresses the file as a whole separately?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2029225879
##########
csharp/src/Drivers/Apache/Spark/Lz4Utilities.cs:
##########
@@ -0,0 +1,57 @@
+/*
+* 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.
+*/
+
+using System;
+using System.Buffers;
+using System.IO;
+using K4os.Compression.LZ4.Streams;
+
+namespace Apache.Arrow.Adbc.Drivers.Apache.Spark
+{
+ /// <summary>
+ /// Utility class for LZ4 compression/decompression operations.
+ /// </summary>
+ internal static class Lz4Utilities
+ {
+ /// <summary>
+ /// Decompresses LZ4 compressed data into memory.
+ /// </summary>
+ /// <param name="compressedData">The compressed data bytes.</param>
+ /// <returns>A ReadOnlyMemory containing the decompressed data.</returns>
+ /// <exception cref="AdbcException">Thrown when decompression fails.</exception>
+ public static ReadOnlyMemory<byte> DecompressLz4(byte[] compressedData)
+ {
+ try
+ {
+ var outputStream = new MemoryStream();
+ using (var inputStream = new MemoryStream(compressedData))
+ using (var decompressor = LZ4Stream.Decode(inputStream))
+ {
+ decompressor.CopyTo(outputStream);
+ }
+ // Get the underlying buffer and its valid length without copying
+ return new ReadOnlyMemory<byte>(outputStream.GetBuffer(), 0, (int)outputStream.Length);
+ // Note: We're not disposing the outputStream here because we're returning its buffer.
+ // The memory will be reclaimed when the ReadOnlyMemory is no longer referenced.
Review Comment:
I don't know that we necessarily need to remove or change the comment, but this isn't strictly correct. There are a lot of misunderstandings about `IDisposable`, but disposing a `MemoryStream` does nothing to its internal buffer. Because the buffer is a managed array, it will not get garbage-collected until there are no more references to it -- even if the `MemoryStream` that created it is disposed.
##########
csharp/src/Drivers/Apache/Spark/Lz4Utilities.cs:
##########
@@ -0,0 +1,57 @@
+/*
+* 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.
+*/
+
+using System;
+using System.Buffers;
+using System.IO;
+using K4os.Compression.LZ4.Streams;
+
+namespace Apache.Arrow.Adbc.Drivers.Apache.Spark
+{
+ /// <summary>
+ /// Utility class for LZ4 compression/decompression operations.
+ /// </summary>
+ internal static class Lz4Utilities
+ {
+ /// <summary>
+ /// Decompresses LZ4 compressed data into memory.
+ /// </summary>
+ /// <param name="compressedData">The compressed data bytes.</param>
+ /// <returns>A ReadOnlyMemory containing the decompressed data.</returns>
+ /// <exception cref="AdbcException">Thrown when decompression fails.</exception>
+ public static ReadOnlyMemory<byte> DecompressLz4(byte[] compressedData)
+ {
+ try
+ {
+ var outputStream = new MemoryStream();
+ using (var inputStream = new MemoryStream(compressedData))
+ using (var decompressor = LZ4Stream.Decode(inputStream))
+ {
+ decompressor.CopyTo(outputStream);
+ }
+ // Get the underlying buffer and its valid length without copying
+ return new ReadOnlyMemory<byte>(outputStream.GetBuffer(), 0, (int)outputStream.Length);
+ // Note: We're not disposing the outputStream here because we're returning its buffer.
+ // The memory will be reclaimed when the ReadOnlyMemory is no longer referenced.
Review Comment:
It's fair to not want to assume anything about what `MemoryStream.Dispose` might do to its internal buffer, but we're now assuming instead that a `MemoryStream` doesn't need to be disposed. (This happens to be true.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "CurtHagenlocher (via GitHub)" <gi...@apache.org>.
CurtHagenlocher commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027111358
##########
csharp/src/Drivers/Apache/Spark/Lz4Utilities.cs:
##########
@@ -0,0 +1,57 @@
+/*
+* 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.
+*/
+
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using K4os.Compression.LZ4.Streams;
+
+namespace Apache.Arrow.Adbc.Drivers.Apache.Spark
+{
+ /// <summary>
+ /// Utility class for LZ4 compression/decompression operations.
+ /// </summary>
+ internal static class Lz4Utilities
+ {
+ /// <summary>
+ /// Decompresses LZ4 compressed data into a memory stream.
+ /// </summary>
+ /// <param name="compressedData">The compressed data bytes.</param>
+ /// <param name="cancellationToken">A cancellation token.</param>
+ /// <returns>A memory stream containing the decompressed data, positioned at the beginning of the stream.</returns>
+ /// <exception cref="AdbcException">Thrown when decompression fails.</exception>
+ public static async Task<MemoryStream> DecompressLz4Async(byte[] compressedData, CancellationToken cancellationToken = default)
+ {
+ try
+ {
+ var outputStream = new MemoryStream();
+ using (var inputStream = new MemoryStream(compressedData))
+ using (var decompressor = LZ4Stream.Decode(inputStream))
+ {
+ await decompressor.CopyToAsync(outputStream);
+ }
+ outputStream.Position = 0;
+ return outputStream;
+ }
+ catch (Exception ex)
+ {
+ throw new AdbcException($"Failed to decompress LZ4 data: {ex.Message}", ex);
+ }
+ }
+ }
+}
Review Comment:
There's a single trailing space on this last line which is triggering the formatting error.
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
+ dataToUse = dataStream.ToArray();
Review Comment:
This could be optimized to remove one allocation and one buffer copy by doing the following:
1) Make `dataToUse` be a `ReadOnlyMemory<byte>` instead of an array.
2) After computing `dataStream`, do `dataToUse = new ReadOnlyMemory<byte>(dataStream.GetBuffer(), 0, dataStream.Length);`
3) Change ChunkStream to accept a `ReadOnlyMemory<byte>` instead of an array as an argument.
If we know the uncompressed length in advance we could do even better. (One of the benefits of the Arrow compressed formats is that it reports that in the header -- but we obviously can't use that format if the server isn't producing it... .)
##########
csharp/src/Drivers/Apache/Spark/Lz4Utilities.cs:
##########
@@ -0,0 +1,57 @@
+/*
+* 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.
+*/
+
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using K4os.Compression.LZ4.Streams;
+
+namespace Apache.Arrow.Adbc.Drivers.Apache.Spark
+{
+ /// <summary>
+ /// Utility class for LZ4 compression/decompression operations.
+ /// </summary>
+ internal static class Lz4Utilities
+ {
+ /// <summary>
+ /// Decompresses LZ4 compressed data into a memory stream.
+ /// </summary>
+ /// <param name="compressedData">The compressed data bytes.</param>
+ /// <param name="cancellationToken">A cancellation token.</param>
+ /// <returns>A memory stream containing the decompressed data, positioned at the beginning of the stream.</returns>
+ /// <exception cref="AdbcException">Thrown when decompression fails.</exception>
+ public static async Task<MemoryStream> DecompressLz4Async(byte[] compressedData, CancellationToken cancellationToken = default)
Review Comment:
FWIW, as this is entirely CPU-bound there's no advantage (and maybe even a small perf hit) in making it `async`.
##########
csharp/src/Drivers/Apache/Spark/SparkStatement.cs:
##########
@@ -155,9 +155,10 @@ internal void SetMaxBytesPerFile(long maxBytesPerFile)
/// </summary>
public sealed class Options : ApacheParameters
Review Comment:
Unrelated to this change, it would be nice if these properties were listed with the others in csharp/src/Drivers/Apache/Spark/README.md. (That could be done as a followup.)
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
+ dataToUse = dataStream.ToArray();
+ dataStream.Dispose();
+ }
+ catch (Exception ex)
+ {
+ // If decompression fails, use the original data
+ System.Diagnostics.Debug.WriteLine($"Failed to decompress LZ4 data: {ex.Message}");
+ }
+ }
+
+ // Always use ChunkStream which ensures proper schema handling
+ this.reader = new ArrowStreamReader(new ChunkStream(this.schema, dataToUse));
+ }
+ catch (Exception ex)
+ {
+ // Log any errors and skip this batch
+ System.Diagnostics.Debug.WriteLine($"Error processing batch: {ex.Message}");
Review Comment:
It seems like a bad idea to just suppress the error. Are there circumstances under which dropping these rows do not result in silent data loss?
##########
csharp/src/Drivers/Apache/Spark/SparkStatement.cs:
##########
@@ -155,9 +155,10 @@ internal void SetMaxBytesPerFile(long maxBytesPerFile)
/// </summary>
public sealed class Options : ApacheParameters
{
+ // Lz4 compression option
+ public const string CanDecompressLz4 = "adbc.spark.lz4Compression.enabled";
Review Comment:
option names in ADBC tend heavily to "snake case" instead of "camel case" e.g. `"lz4_compression"` vs `"lz4Compression"`.
##########
csharp/src/Drivers/Apache/Spark/SparkDatabricksReader.cs:
##########
@@ -79,6 +91,49 @@ public SparkDatabricksReader(HiveServer2Statement statement, Schema schema)
}
}
+ private async Task ProcessFetchedBatchesAsync(CancellationToken cancellationToken)
+ {
+ var batch = this.batches![this.index];
+
+ // Ensure batch data exists
+ if (batch.Batch == null || batch.Batch.Length == 0)
+ {
+ this.index++;
+ return;
+ }
+
+ try
+ {
+ byte[] dataToUse = batch.Batch;
+
+ // If LZ4 compression is enabled, try to decompress the data
+ if (isLz4Compressed)
+ {
+ try
+ {
+ var dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
Review Comment:
```suggestion
using dataStream = await Lz4Utilities.DecompressLz4Async(batch.Batch, cancellationToken);
```
Then the explicit `Dispose()` can be removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(csharp/src/Drivers/Apache/Spark): Add Lz4 compression support to arrow batch reader [arrow-adbc]
Posted by "eric-wang-1990 (via GitHub)" <gi...@apache.org>.
eric-wang-1990 commented on code in PR #2669:
URL: https://github.com/apache/arrow-adbc/pull/2669#discussion_r2027760168
##########
csharp/src/Drivers/Apache/Spark/SparkStatement.cs:
##########
@@ -155,9 +155,10 @@ internal void SetMaxBytesPerFile(long maxBytesPerFile)
/// </summary>
public sealed class Options : ApacheParameters
Review Comment:
These params are statement level for now, I will create a follow up to change it to connection level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org