viirya commented on code in PR #6883: URL: https://github.com/apache/datafusion-comet/pull/6883#discussion_r4432795518
########## spark/src/main/scala/org/apache/comet/local/CometLocalResultExec.scala: ########## @@ -0,0 +1,128 @@ +/* + * 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. + */ + +package org.apache.comet.local + +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong + +import scala.collection.mutable.ArrayBuffer + +import org.apache.spark.SparkException +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder, UnsafeRow} +import org.apache.spark.sql.catalyst.plans.physical.Partitioning +import org.apache.spark.sql.execution.{ColumnarToRowExec, SparkPlan, UnaryExecNode} + +/** + * Row result boundary of a local query. Spark's collect encodes and compresses every row of each + * result partition in its task, then decodes them on the driver. A local collect instead runs the + * whole query as one native graph in a single Spark task (keeping cancellation, job groups and + * SQL metrics), which hands copied rows to the driver in this JVM, since admission requires an + * in-process local master. + * + * Only `collect` runs that way. Everything else reads the result as an RDD: a take requests + * partitions in growing batches, an iterator fetches one partition at a time, a tail scans from + * the last one, and a cache, `df.rdd` or PySpark's collect keep its partitions. So the node + * executes as the ordinary native block otherwise, one Spark task per partition, and reports the + * block's partitioning and ordering. + */ +case class CometLocalResultExec(child: SparkPlan) extends UnaryExecNode { + override def output: Seq[Attribute] = child.output + override def outputPartitioning: Partitioning = ordinary.outputPartitioning + override def outputOrdering: Seq[SortOrder] = ordinary.outputOrdering + + override protected def doExecute(): RDD[InternalRow] = ordinary.execute() + + override def executeCollect(): Array[InternalRow] = + if (sparkContext.isLocal) LocalResultHandoff.collect(child.execute(), maxResultSize) + else super.executeCollect() + + private def ordinary: SparkPlan = + child + .collectFirst { case local: CometLocalExec => ColumnarToRowExec(local.block) } + .getOrElse(child) + + // Spark enforces this on serialized task results; this boundary enforces it while copying. + private def maxResultSize: Long = + sparkContext.getConf.getSizeAsBytes("spark.driver.maxResultSize", "1g") + + override protected def withNewChildInternal(newChild: SparkPlan): CometLocalResultExec = + copy(child = newChild) +} + +/** Driver-owned result slots, keyed by numeric IDs that the task closure carries. */ +private[local] object LocalResultHandoff { + private final class Slot(val maxResultSize: Long) { + @volatile var rows: Array[InternalRow] = _ + @volatile var exceededBytes: Long = -1L + } + + private val slots = new ConcurrentHashMap[Long, Slot]() + private val nextId = new AtomicLong() + + /** Number of live slots; zero whenever no collect is running. */ + def active: Int = slots.size() + + /** + * Runs the single result partition of `rdd` and returns its rows. Stops reading, which closes + * the native query, once the copied rows exceed `maxResultSize` bytes (unlimited if not + * positive). Unlike Spark's compressed size, this counts uncompressed UnsafeRow bytes, so it is + * the more conservative of the two. + */ + def collect(rdd: RDD[InternalRow], maxResultSize: Long): Array[InternalRow] = { + require(rdd.getNumPartitions == 1, "Local results have exactly one partition") + val id = nextId.incrementAndGet() + val slot = new Slot(maxResultSize) + slots.put(id, slot) + try { + rdd.sparkContext.runJob(rdd, (rows: Iterator[InternalRow]) => fill(id, rows), Seq(0)) + if (slot.exceededBytes >= 0) { + throw new SparkException( + s"Total size of local query results (at least ${slot.exceededBytes} bytes, " + + s"uncompressed) is bigger than spark.driver.maxResultSize ($maxResultSize bytes)") + } + Option(slot.rows).getOrElse(throw new IllegalStateException("Local result not delivered")) + } finally { + slots.remove(id) + () + } + } + + // Runs in the result task. A retried attempt rebuilds the result and publishes it only once + // complete. A slot the driver already removed means nobody will read the rows. + private def fill(id: Long, rows: Iterator[InternalRow]): Unit = { + val slot = slots.get(id) + if (slot == null) return Review Comment: Fixed in f99686ccd. Admission now declines when `spark.executor.userClassPathFirst` is set, since the task's loader would load its own copy of the result slots. A registry shared across loaders would still need a slot type that both loaders resolve to the same class, so declining seemed the safer choice. ########## native/core/src/local.rs: ########## @@ -0,0 +1,259 @@ +// 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. + +//! JNI adapter for local execution, which runs a whole admitted Comet plan as one DataFusion +//! graph inside the result task of a local Spark application. The registry holds only live +//! queries, never plans for reuse. EOF, errors and explicit close remove entries; Java task +//! completion also closes. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; + +use datafusion::physical_plan::union::UnionExec; +use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_comet_jni_bridge::{jni_new_global_ref, JVMClasses}; +use datafusion_comet_local::handle::{QueryHandle, QueryPoll}; +use datafusion_comet_local::LocalQuery; +use jni::objects::{JByteArray, JLongArray, JObject, JObjectArray, JString, ReleaseMode}; +use jni::sys::{jboolean, jint, jlong, JNI_FALSE}; +use jni::{Env, EnvUnowned}; +use parking_lot::Mutex; + +use crate::errors::{try_unwrap_or_throw, CometError, CometResult}; +use crate::execution::jni_api::{ + create_execution_context, create_memory_pool, get_runtime, prepare_output, + prepare_plan_settings, ExecutionContext, PlanInputs, PlanJvmRefs, +}; +use crate::execution::spark_config::{SparkConfig, SPARK_EXECUTOR_CORES}; + +struct Entry { + query: QueryHandle, + columns: usize, + // Holds the plan's memory pool and session until the query is closed. Dropped after the + // query, which cancels the graph that reserves from the pool. Only the creating thread + // uses it; the mutex makes the entry shareable with result pulls. + _context: Mutex<Box<ExecutionContext>>, +} + +// JNI carries numeric IDs instead of raw pointers, so close racing with a result pull +// cannot free an object the pull is using. Last reader releases the removed entry. +static QUERIES: LazyLock<Mutex<HashMap<i64, Arc<Entry>>>> = LazyLock::new(Mutex::default); +static NEXT_ID: AtomicI64 = AtomicI64::new(1); +const MAX_LIVE_QUERIES: usize = 1024; + +fn close(id: i64) { + let entry = QUERIES.lock().remove(&id); + if let Some(entry) = entry { + entry.query.cancel(); + } +} + +/// Creates and starts a local query over `partitions` partitions of a serialized Comet plan +/// without JVM inputs. Its memory pool and DataFusion session are those `Native.createPlan` +/// creates for a plan of the calling Spark task. +#[no_mangle] +pub extern "system" fn Java_org_apache_comet_local_NativeLocal_createQuery( + e: EnvUnowned, + _: JObject, + serialized_query: JByteArray, + serialized_spark_configs: JByteArray, + partitions: jint, + columns: jint, + comet_task_memory_manager_obj: JObject, + local_dirs: JObjectArray, + batch_size: jint, + off_heap_mode: jboolean, + memory_pool_type: JString, + memory_limit: jlong, + off_heap_size: jlong, + task_attempt_id: jlong, + task_cpus: jlong, + read_ahead_bytes: jlong, +) -> jlong { + try_unwrap_or_throw(&e, |env| { + if !(1..=MAX_PARTITIONS).contains(&partitions) || columns < 1 { + return Err(CometError::Internal( + "Invalid local query parameters".into(), + )); + } + let start = Instant::now(); + let settings = prepare_plan_settings(&env.convert_byte_array(serialized_spark_configs)?)?; + // Run as many partitions at once as Spark would run tasks, which also bounds how many + // operators share the memory pool. `local[N]` resolves to N here. + let max_running = settings.spark_config.get_usize(SPARK_EXECUTOR_CORES, 1); Review Comment: Fixed in f99686ccd. Slots are now `spark.executor.cores / spark.task.cpus`, as `LocalSchedulerBackend` counts them, so `local[32]` with `spark.task.cpus=16` gives 2. ########## native/core/src/local.rs: ########## @@ -0,0 +1,259 @@ +// 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. + +//! JNI adapter for local execution, which runs a whole admitted Comet plan as one DataFusion +//! graph inside the result task of a local Spark application. The registry holds only live +//! queries, never plans for reuse. EOF, errors and explicit close remove entries; Java task +//! completion also closes. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; + +use datafusion::physical_plan::union::UnionExec; +use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_comet_jni_bridge::{jni_new_global_ref, JVMClasses}; +use datafusion_comet_local::handle::{QueryHandle, QueryPoll}; +use datafusion_comet_local::LocalQuery; +use jni::objects::{JByteArray, JLongArray, JObject, JObjectArray, JString, ReleaseMode}; +use jni::sys::{jboolean, jint, jlong, JNI_FALSE}; +use jni::{Env, EnvUnowned}; +use parking_lot::Mutex; + +use crate::errors::{try_unwrap_or_throw, CometError, CometResult}; +use crate::execution::jni_api::{ + create_execution_context, create_memory_pool, get_runtime, prepare_output, + prepare_plan_settings, ExecutionContext, PlanInputs, PlanJvmRefs, +}; +use crate::execution::spark_config::{SparkConfig, SPARK_EXECUTOR_CORES}; + +struct Entry { + query: QueryHandle, + columns: usize, + // Holds the plan's memory pool and session until the query is closed. Dropped after the + // query, which cancels the graph that reserves from the pool. Only the creating thread + // uses it; the mutex makes the entry shareable with result pulls. + _context: Mutex<Box<ExecutionContext>>, +} + +// JNI carries numeric IDs instead of raw pointers, so close racing with a result pull +// cannot free an object the pull is using. Last reader releases the removed entry. +static QUERIES: LazyLock<Mutex<HashMap<i64, Arc<Entry>>>> = LazyLock::new(Mutex::default); +static NEXT_ID: AtomicI64 = AtomicI64::new(1); +const MAX_LIVE_QUERIES: usize = 1024; + +fn close(id: i64) { + let entry = QUERIES.lock().remove(&id); + if let Some(entry) = entry { + entry.query.cancel(); Review Comment: Fixed in f99686ccd. `Entry` now cancels the query and then releases the context through `release_plan`, so both teardown paths share the bounded `plan_memory` wait, including early exits on cancellation or the result-size limit. ########## spark/src/main/scala/org/apache/comet/local/CometLocalResultExec.scala: ########## @@ -0,0 +1,128 @@ +/* + * 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. + */ + +package org.apache.comet.local + +import java.util.concurrent.ConcurrentHashMap +import java.util.concurrent.atomic.AtomicLong + +import scala.collection.mutable.ArrayBuffer + +import org.apache.spark.SparkException +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder, UnsafeRow} +import org.apache.spark.sql.catalyst.plans.physical.Partitioning +import org.apache.spark.sql.execution.{ColumnarToRowExec, SparkPlan, UnaryExecNode} + +/** + * Row result boundary of a local query. Spark's collect encodes and compresses every row of each + * result partition in its task, then decodes them on the driver. A local collect instead runs the + * whole query as one native graph in a single Spark task (keeping cancellation, job groups and + * SQL metrics), which hands copied rows to the driver in this JVM, since admission requires an + * in-process local master. + * + * Only `collect` runs that way. Everything else reads the result as an RDD: a take requests + * partitions in growing batches, an iterator fetches one partition at a time, a tail scans from + * the last one, and a cache, `df.rdd` or PySpark's collect keep its partitions. So the node + * executes as the ordinary native block otherwise, one Spark task per partition, and reports the + * block's partitioning and ordering. + */ +case class CometLocalResultExec(child: SparkPlan) extends UnaryExecNode { + override def output: Seq[Attribute] = child.output + override def outputPartitioning: Partitioning = ordinary.outputPartitioning + override def outputOrdering: Seq[SortOrder] = ordinary.outputOrdering + + override protected def doExecute(): RDD[InternalRow] = ordinary.execute() + + override def executeCollect(): Array[InternalRow] = + if (sparkContext.isLocal) LocalResultHandoff.collect(child.execute(), maxResultSize) + else super.executeCollect() + + private def ordinary: SparkPlan = + child + .collectFirst { case local: CometLocalExec => ColumnarToRowExec(local.block) } Review Comment: Fixed in f99686ccd. `ordinary` now applies `EliminateRedundantTransitions` to `ColumnarToRowExec(block)`, so a bare range runs as Spark's `RangeExec` and a projection gets Comet's columnar-to-row transition. The RDD test checks both. ########## native/core/src/local.rs: ########## @@ -0,0 +1,259 @@ +// 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. + +//! JNI adapter for local execution, which runs a whole admitted Comet plan as one DataFusion +//! graph inside the result task of a local Spark application. The registry holds only live +//! queries, never plans for reuse. EOF, errors and explicit close remove entries; Java task +//! completion also closes. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; + +use datafusion::physical_plan::union::UnionExec; +use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties}; +use datafusion_comet_jni_bridge::{jni_new_global_ref, JVMClasses}; +use datafusion_comet_local::handle::{QueryHandle, QueryPoll}; +use datafusion_comet_local::LocalQuery; +use jni::objects::{JByteArray, JLongArray, JObject, JObjectArray, JString, ReleaseMode}; +use jni::sys::{jboolean, jint, jlong, JNI_FALSE}; +use jni::{Env, EnvUnowned}; +use parking_lot::Mutex; + +use crate::errors::{try_unwrap_or_throw, CometError, CometResult}; +use crate::execution::jni_api::{ + create_execution_context, create_memory_pool, get_runtime, prepare_output, + prepare_plan_settings, ExecutionContext, PlanInputs, PlanJvmRefs, +}; +use crate::execution::spark_config::{SparkConfig, SPARK_EXECUTOR_CORES}; + +struct Entry { + query: QueryHandle, + columns: usize, + // Holds the plan's memory pool and session until the query is closed. Dropped after the + // query, which cancels the graph that reserves from the pool. Only the creating thread + // uses it; the mutex makes the entry shareable with result pulls. + _context: Mutex<Box<ExecutionContext>>, +} + +// JNI carries numeric IDs instead of raw pointers, so close racing with a result pull +// cannot free an object the pull is using. Last reader releases the removed entry. +static QUERIES: LazyLock<Mutex<HashMap<i64, Arc<Entry>>>> = LazyLock::new(Mutex::default); +static NEXT_ID: AtomicI64 = AtomicI64::new(1); +const MAX_LIVE_QUERIES: usize = 1024; + +fn close(id: i64) { + let entry = QUERIES.lock().remove(&id); + if let Some(entry) = entry { + entry.query.cancel(); + } +} + +/// Creates and starts a local query over `partitions` partitions of a serialized Comet plan +/// without JVM inputs. Its memory pool and DataFusion session are those `Native.createPlan` +/// creates for a plan of the calling Spark task. +#[no_mangle] +pub extern "system" fn Java_org_apache_comet_local_NativeLocal_createQuery( + e: EnvUnowned, + _: JObject, + serialized_query: JByteArray, + serialized_spark_configs: JByteArray, + partitions: jint, + columns: jint, + comet_task_memory_manager_obj: JObject, + local_dirs: JObjectArray, + batch_size: jint, + off_heap_mode: jboolean, + memory_pool_type: JString, + memory_limit: jlong, + off_heap_size: jlong, + task_attempt_id: jlong, + task_cpus: jlong, + read_ahead_bytes: jlong, +) -> jlong { + try_unwrap_or_throw(&e, |env| { + if !(1..=MAX_PARTITIONS).contains(&partitions) || columns < 1 { + return Err(CometError::Internal( + "Invalid local query parameters".into(), + )); + } + let start = Instant::now(); + let settings = prepare_plan_settings(&env.convert_byte_array(serialized_spark_configs)?)?; + // Run as many partitions at once as Spark would run tasks, which also bounds how many + // operators share the memory pool. `local[N]` resolves to N here. + let max_running = settings.spark_config.get_usize(SPARK_EXECUTOR_CORES, 1); + JVMClasses::init(env); + let serialized_plan = env.convert_byte_array(serialized_query)?; + let task_memory_manager = + Arc::new(jni_new_global_ref!(env, comet_task_memory_manager_obj)?); + let memory_pool_type = memory_pool_type.try_to_string(env)?; + let mut local_dirs_vec = vec![]; + for i in 0..local_dirs.len(env)? { + let dir = local_dirs.get_element(env, i)?; + let dir = unsafe { JString::from_raw(&*env, dir.into_raw()) }; + local_dirs_vec.push(dir.try_to_string(env)?); + } + let id = next_id()?; + let inputs = PlanInputs { + id, + settings, + serialized_plan, + partition_count: partitions as usize, + batch_size: batch_size as usize, + off_heap_mode: off_heap_mode != JNI_FALSE, + memory_pool_type, + memory_limit, + off_heap_size, + task_attempt_id, + task_cpus: task_cpus as usize, + local_dirs: local_dirs_vec, + metrics_update_interval: 0, + start, + jvm: PlanJvmRefs::default(), + }; + let context = create_execution_context(inputs, |config| { + create_memory_pool(config, task_memory_manager, task_attempt_id) + })?; + let plan = local_plan(&context, partitions)?; + if plan.schema().fields().len() != columns as usize { + return Err(CometError::Internal( + "Local output schema width mismatch".into(), + )); + } + let query = LocalQuery::new(plan, context.session_ctx.task_ctx()) + .in_partition_order(max_running, read_ahead_bytes.max(0) as usize); + let query = QueryHandle::start(query, &get_runtime())?; + register( + id, + Entry { + query, + columns: columns as usize, + _context: Mutex::new(context), + }, + )?; + Ok(id) + }) +} + +const MAX_PARTITIONS: jint = 1024; + +/// One graph for the whole query. Each partition's operators are created by the planner for +/// that partition, as a Spark task would create them, so per-partition state such as a range +/// slice or a partition-dependent expression is correct. Partition `i` of the graph is the +/// single output partition of the operators created for partition `i`. +fn local_plan(context: &ExecutionContext, partitions: jint) -> CometResult<Arc<dyn ExecutionPlan>> { + let mut plans = Vec::with_capacity(partitions as usize); + for partition in 0..partitions { + let (scans, shuffle_scans, root) = context.planner(partition).create_plan( + &context.spark_plan, + &mut vec![], + partitions as usize, + )?; + if !scans.is_empty() + || !shuffle_scans.is_empty() + || root.native_plan.output_partitioning().partition_count() != 1 + { + return Err(CometError::Internal( + "Local query plans must have no JVM input and one output partition".into(), + )); + } + plans.push(Arc::clone(&root.native_plan)); + } + Ok(UnionExec::try_new(plans)?) +} + +fn next_id() -> CometResult<i64> { + // IDs are never reused. A counter that wrapped would reuse them, so fail instead. + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + if id < 1 { + return Err(CometError::Internal("Local query IDs exhausted".into())); + } + Ok(id) +} + +fn register(id: i64, entry: Entry) -> CometResult<()> { + let mut registry = QUERIES.lock(); + if registry.len() >= MAX_LIVE_QUERIES { + return Err(CometError::Internal("Too many live local queries".into())); + } + registry.insert(id, Arc::new(entry)); + Ok(()) +} + +fn next(env: &mut Env, id: i64, arrays: JLongArray, schemas: JLongArray) -> CometResult<jlong> { + let entry = QUERIES + .lock() + .get(&id) + .cloned() + .ok_or_else(|| CometError::Internal("Local query is closed or unknown".into()))?; + if arrays.len(env)? != entry.columns || schemas.len(env)? != entry.columns { + return Err(CometError::Internal( + "Local output column count mismatch".into(), + )); + } + // The caller supplies writable Arrow C structs. Reject null pointers before exporting. + for addresses in [&arrays, &schemas] { + let elements = unsafe { addresses.get_elements(env, ReleaseMode::NoCopyBack)? }; + if elements.contains(&0) { + return Err(CometError::Internal("Null local output pointer".into())); + } + } + match entry.query.poll(Duration::from_millis(50))? { + QueryPoll::Pending => Ok(-2), + QueryPoll::Finished => Ok(-1), + QueryPoll::Batch(batch) => prepare_output(env, arrays, schemas, batch, false), Review Comment: Fixed in f99686ccd. The JNI pull passes the context's `debug_native` to `prepare_output`. `LocalQuerySpec` now carries the output types, and `LocalQueryIterator` runs `checkTimestampLabels` when strict labels are enabled. A test runs a timestamp projection with both switches on. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
