Fix compile issues

Author: m09526

java/compaction/compaction-core/src/test/java/sleeper/compaction/core/task/CompactionTaskTestBase.java

@@ -154,7 +154,7 @@ private void runTask(
         CompactionJobCommitterOrSendToLambda committer = new CompactionJobCommitterOrSendToLambda(
                 tablePropertiesProvider(), stateStoreProvider(),
                 jobTracker, stateStoreCommitQueue::add, batcherCommitQueue::add, timeSupplier);
-        CompactionRunnerFactory selector = (job, properties) -> compactor;
+        CompactionRunnerFactory selector = (job, instanceProperties, properties) -> compactor;
         new CompactionTask(instanceProperties, tablePropertiesProvider(), PropertiesReloader.neverReload(),
                 stateStoreProvider(), messageReceiver, fileAssignmentCheck,
                 committer, jobTracker, taskTracker, selector, taskId, jobRunIdSupplier, timeSupplier, sleeps::add)

java/compaction/compaction-gpu/pom.xml

@@ -20,7 +20,7 @@
     <parent>
         <artifactId>compaction</artifactId>
         <groupId>sleeper</groupId>
-        <version>0.28.0-SNAPSHOT</version>
+        <version>0.29.0-SNAPSHOT</version>
     </parent>
     <properties>
         <protobuf.version>3.25.1</protobuf.version>

java/compaction/compaction-gpu/src/main/java/sleeper/compaction/gpu/GPUCompactionRunner.java

@@ -34,13 +34,13 @@
 import sleeper.core.properties.table.TableProperties;
 import sleeper.core.range.Range;
 import sleeper.core.range.Region;
-import sleeper.core.record.process.RecordsProcessed;
 import sleeper.core.schema.Schema;
 import sleeper.core.schema.type.ByteArrayType;
 import sleeper.core.schema.type.IntType;
 import sleeper.core.schema.type.LongType;
 import sleeper.core.schema.type.PrimitiveType;
 import sleeper.core.schema.type.StringType;
+import sleeper.core.tracker.job.run.RecordsProcessed;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;

rust/compaction/src/datafusion.rs

@@ -82,9 +82,14 @@ pub async fn compact(
         })
         .inspect_err(|e| warn!("Error getting total input size {e}"));
     let multipart_size = std::cmp::max(
+        crate::store::MULTIPART_BUF_SIZE,
+        input_size.unwrap_or_default() / 5000,
+    );
+    store_factory
         .get_object_store(output_path)
         .map_err(|e| DataFusionError::External(e.into()))?
         .set_multipart_size_hint(multipart_size);
+    info!(
         "Setting multipart size hint to {} bytes.",
         multipart_size.to_formatted_string(&Locale::en)
     );

rust/compaction/src/datafusion/udf.rs

@@ -19,6 +19,13 @@
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
+use std::{
+    any::Any,
+    fmt::Debug,
+    iter::zip,
+    sync::{Mutex, MutexGuard},
+};
+
 use arrow::{
     array::AsArray,
     datatypes::{
@@ -31,16 +38,6 @@ use datafusion::{
     prelude::Expr,
     scalar::ScalarValue,
 };
-use log::info;
-use num_format::{Locale, ToFormattedString};
-
-use std::{
-    any::Any,
-    fmt::Debug,
-    iter::zip,
-    sync::{Mutex, MutexGuard},
-    time::{Duration, Instant},
-};
 
 use super::sketch::{update_sketch, DataSketchVariant, K};
 
@@ -50,12 +47,10 @@ use super::sketch::{update_sketch, DataSketchVariant, K};
 /// so we can inject them into a sketch for later retrieval. The query should look something like:
 /// `SELECT sketch(row_key_col1, row_key_col2, ...), row_key_col2, value_col1, value_col2, ... FROM blah...`
 /// so the sketch function can see each row key column, but only returns the first.
-
 pub(crate) struct SketchUDF {
     signature: Signature,
     invoke_count: Mutex<usize>,
     sketch: Mutex<Vec<DataSketchVariant>>,
-    bench_stats: Mutex<BenchStats>,
 }
 
 impl Debug for SketchUDF {
@@ -69,21 +64,13 @@ impl Debug for SketchUDF {
 }
 
 impl SketchUDF {
-    const BENCH_DURATION: Duration = Duration::from_secs(10);
     /// Create a new sketch function based on the schema of the row key fields.
     ///
     pub fn new(schema: &DFSchema, row_keys: &[String]) -> Self {
         Self {
             signature: Signature::exact(get_row_key_types(schema, row_keys), Volatility::Immutable),
             invoke_count: Mutex::default(),
             sketch: Mutex::new(make_sketches_for_schema(schema, row_keys)),
-            bench_stats: Mutex::new(BenchStats {
-                start_time: Instant::now(),
-                last_instant_measure: Instant::now(),
-                last_instant_row_speed: 0,
-                rows_since_instant: 0,
-                rows_since_start: 0,
-            }),
         }
     }
 
@@ -94,35 +81,6 @@ impl SketchUDF {
     pub fn get_invoke_count(&self) -> usize {
         *self.invoke_count.lock().unwrap()
     }
-
-    fn bench_report(&self, rows: usize) {
-        let mut stats_lock = self.bench_stats.lock().unwrap();
-        stats_lock.rows_since_start += rows;
-        stats_lock.rows_since_instant += rows;
-        let now = Instant::now();
-
-        if now.duration_since(stats_lock.last_instant_measure) > Self::BENCH_DURATION {
-            stats_lock.last_instant_row_speed = stats_lock.rows_since_instant as u64
-                / std::cmp::max(
-                    now.duration_since(stats_lock.last_instant_measure)
-                        .as_secs(),
-                    1,
-                );
-            stats_lock.last_instant_measure = now;
-            stats_lock.rows_since_instant = 0;
-            let rows_speed_from_start = stats_lock.rows_since_start as u64
-                / std::cmp::max(now.duration_since(stats_lock.start_time).as_secs(), 1);
-
-            info!(
-                "Bench speeds: {} rows/sec. ({} second rolling avg.) {} rows/sec. lifetime avg.",
-                stats_lock
-                    .last_instant_row_speed
-                    .to_formatted_string(&Locale::en),
-                Self::BENCH_DURATION.as_secs(),
-                rows_speed_from_start.to_formatted_string(&Locale::en)
-            );
-        }
-    }
 }
 
 /// Create a [`Vec`] of data types for this schema from the row keys.
@@ -191,7 +149,6 @@ impl ScalarUDFImpl for SketchUDF {
         for (sketch, col) in zip(sk_lock.iter_mut(), &args.args) {
             match col {
                 ColumnarValue::Array(array) => {
-                    self.bench_report(array.len());
                     // dynamic dispatch. Match the datatype to the type of sketch to update.
                     match array.data_type() {
                         DataType::Int32 => update_sketch(sketch, &array.as_primitive::<Int32Type>()),
@@ -229,23 +186,19 @@ impl ScalarUDFImpl for SketchUDF {
                     | ScalarValue::LargeUtf8(Some(value))
                     | ScalarValue::Utf8View(Some(value)),
                 ) => {
-                    self.bench_report(1);
                     sketch.update(value);
                 }
                 ColumnarValue::Scalar(
                     ScalarValue::Binary(Some(value))
                     | ScalarValue::LargeBinary(Some(value))
                     | ScalarValue::BinaryView(Some(value)),
                 ) => {
-                    self.bench_report(1);
                     sketch.update(value);
                 }
                 ColumnarValue::Scalar(ScalarValue::Int32(Some(value))) => {
-                    self.bench_report(1);
                     sketch.update(value);
                 }
                 ColumnarValue::Scalar(ScalarValue::Int64(Some(value))) => {
-                    self.bench_report(1);
                     sketch.update(value);
                 }
                 x @ ColumnarValue::Scalar(_) => {