chore: Remove fast encoding option (#1703)

Author: andygrove

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -326,14 +326,6 @@ object CometConf extends ShimCometConf {
       .intConf
       .createWithDefault(1)
 
-  val COMET_SHUFFLE_ENABLE_FAST_ENCODING: ConfigEntry[Boolean] =
-    conf(s"$COMET_EXEC_CONFIG_PREFIX.shuffle.enableFastEncoding")
-      .doc("Whether to enable Comet's faster proprietary encoding for shuffle blocks " +
-        "rather than using Arrow IPC.")
-      .internal()
-      .booleanConf
-      .createWithDefault(true)
-
   val COMET_COLUMNAR_SHUFFLE_ASYNC_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.columnar.shuffle.async.enabled")
       .doc("Whether to enable asynchronous shuffle for Arrow-based shuffle.")

native/core/benches/row_columnar.rs

@@ -79,7 +79,6 @@ fn benchmark(c: &mut Criterion) {
                 0,
                 None,
                 &CompressionCodec::Zstd(1),
-                true,
             )
             .unwrap();
         });

native/core/benches/shuffle_writer.rs

@@ -47,12 +47,8 @@ fn criterion_benchmark(c: &mut Criterion) {
             group.bench_function(name, |b| {
                 let mut buffer = vec![];
                 let ipc_time = Time::default();
-                let w = ShuffleBlockWriter::try_new(
-                    &batch.schema(),
-                    enable_fast_encoding,
-                    compression_codec.clone(),
-                )
-                .unwrap();
+                let w = ShuffleBlockWriter::try_new(&batch.schema(), compression_codec.clone())
+                    .unwrap();
                 b.iter(|| {
                     buffer.clear();
                     let mut cursor = Cursor::new(&mut buffer);
@@ -97,7 +93,6 @@ fn create_shuffle_writer_exec(compression_codec: CompressionCodec) -> ShuffleWri
         compression_codec,
         "/tmp/data.out".to_string(),
         "/tmp/index.out".to_string(),
-        true,
     )
     .unwrap()
 }

native/core/src/execution/jni_api.rs

@@ -537,7 +537,6 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_writeSortedFileNative
     current_checksum: jlong,
     compression_codec: jstring,
     compression_level: jint,
-    enable_fast_encoding: jboolean,
 ) -> jlongArray {
     try_unwrap_or_throw(&e, |mut env| unsafe {
         let data_types = convert_datatype_arrays(&mut env, serialized_datatypes)?;
@@ -589,7 +588,6 @@ pub unsafe extern "system" fn Java_org_apache_comet_Native_writeSortedFileNative
             checksum_algo,
             current_checksum,
             &compression_codec,
-            enable_fast_encoding != JNI_FALSE,
         )?;
 
         let checksum = if let Some(checksum) = checksum {

native/core/src/execution/planner.rs

@@ -1202,7 +1202,6 @@ impl PhysicalPlanner {
                     codec,
                     writer.output_data_file.clone(),
                     writer.output_index_file.clone(),
-                    writer.enable_fast_encoding,
                 )?);
 
                 Ok((