SemyonSinchenko commented on code in PR #58302: URL: https://github.com/apache/spark/pull/58302#discussion_r4272377360
########## graphframes/src/main/scala/org/apache/spark/graphframes/embeddings/Hash2Vec.scala: ########## @@ -0,0 +1,603 @@ +/* + * 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.spark.graphframes.embeddings + +import scala.reflect.ClassTag +import scala.util.hashing.MurmurHash3 + +import dev.ludovic.netlib.blas.BLAS + +import org.apache.spark.graphframes.GraphFramesUnsupportedVertexTypeException +import org.apache.spark.graphframes.rw.RandomWalkBase +import org.apache.spark.ml.linalg +import org.apache.spark.ml.linalg.SQLDataTypes.VectorType +import org.apache.spark.ml.linalg.Vectors +import org.apache.spark.ml.stat.Summarizer +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.Row +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.hash +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.pmod +import org.apache.spark.sql.functions.transform +import org.apache.spark.sql.functions.udf +import org.apache.spark.sql.types.ArrayType +import org.apache.spark.sql.types.ByteType +import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.types.LongType +import org.apache.spark.sql.types.ShortType +import org.apache.spark.sql.types.StringType +import org.apache.spark.sql.types.StructField +import org.apache.spark.sql.types.StructType +import org.apache.spark.unsafe.hash.Murmur3_x86_32._ + +/** + * Implementation of Hash2Vec, an efficient word embedding technique using feature hashing. Based + * on: Argerich, Luis, Joaquin Torre Zaffaroni, and Matias J. Cano. "Hash2vec, feature hashing for + * word embeddings." arXiv preprint arXiv:1608.08940 (2016). + * + * Produces embeddings for elements in sequences using a hash-based approach to avoid storing a + * vocabulary. Uses MurmurHash3 for hashing elements to embedding indices and signs. + * + * Output DataFrame has columns "id" (element identifier, same type as sequence elements) and + * "vector" (dense vector of doubles, summed across all occurrences). + * + * Tradeoffs: Higher numPartitions reduces local state and memory per partition but increases + * aggregation and merging overhead across partitions. Larger embeddingsDim provides richer + * representations but consumes more memory. Seeds control hashing for reproducibility. + */ +class Hash2Vec extends Serializable { Review Comment: What do you think about moving this to the ML? I mean it is an implementation of the (almost) generic seq2vec model that can be scaled beyond the limits if the ML's `word2vec` (hard cap ~20M of words in vocab with default dim=100). I battle-tested it on the corpus with more than 50M of words in vocab and it works. ########## graphframes/src/main/scala/org/apache/spark/graphframes/Logging.scala: ########## @@ -0,0 +1,47 @@ +/* + * 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.spark.graphframes + +import org.slf4j.Logger +import org.slf4j.LoggerFactory + +// This needs to be accessible to org.apache.spark.graphx.lib.backport +private[org] trait Logging { Review Comment: Do we really need it there? From my understanding of the old GF' codebase this was more like a workaround. But from inside of the Apache Spark we can use it's own logging logic instead of duplicating it. ########## graphframes/src/main/scala/org/apache/spark/graphframes/examples/BeliefPropagation.scala: ########## Review Comment: What do you think about moving examples out of the core codebase? We did not do this in GF due to the back-compat policy, but now it is a perfect chance: otherwise we may stuck with this "examples" ########## graphframes/src/test/scala/org/apache/spark/graphframes/TestUtils.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.spark.graphframes + +import org.apache.spark.graphframes.GraphFrame._ +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.types.DataType +import org.apache.spark.sql.types.StructType + +object TestUtils { + + private[this] val majorMinorRegex = """^(\d+)\.(\d+)(\..*)?$""".r + + /** Extract major/minor version integer pairs from a version string */ + def majorMinorVersion(sparkVersion: String): (Int, Int) = { + majorMinorRegex.findFirstMatchIn(sparkVersion) match { + case Some(m) => + (m.group(1).toInt, m.group(2).toInt) + case None => + throw new IllegalArgumentException( + s"Spark tried to parse '$sparkVersion' as a Spark" + + " version string, but it could not find the major and minor version numbers.") + } + } + + /** Return true if the major and minor versions are greater or eq to constraints */ + def requireSparkVersionGE(major: Int, minor: Int, sparkVersion: String): Boolean = { Review Comment: This was a part of version-matrix test, I don't think we really need it. ########## graphframes/src/main/scala/org/apache/spark/graphframes/embeddings/Hash2Vec.scala: ########## @@ -0,0 +1,603 @@ +/* + * 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.spark.graphframes.embeddings + +import scala.reflect.ClassTag +import scala.util.hashing.MurmurHash3 + +import dev.ludovic.netlib.blas.BLAS + +import org.apache.spark.graphframes.GraphFramesUnsupportedVertexTypeException +import org.apache.spark.graphframes.rw.RandomWalkBase +import org.apache.spark.ml.linalg +import org.apache.spark.ml.linalg.SQLDataTypes.VectorType +import org.apache.spark.ml.linalg.Vectors +import org.apache.spark.ml.stat.Summarizer +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.Row +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.hash +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.pmod +import org.apache.spark.sql.functions.transform +import org.apache.spark.sql.functions.udf +import org.apache.spark.sql.types.ArrayType +import org.apache.spark.sql.types.ByteType +import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.types.LongType +import org.apache.spark.sql.types.ShortType +import org.apache.spark.sql.types.StringType +import org.apache.spark.sql.types.StructField +import org.apache.spark.sql.types.StructType +import org.apache.spark.unsafe.hash.Murmur3_x86_32._ + +/** + * Implementation of Hash2Vec, an efficient word embedding technique using feature hashing. Based + * on: Argerich, Luis, Joaquin Torre Zaffaroni, and Matias J. Cano. "Hash2vec, feature hashing for + * word embeddings." arXiv preprint arXiv:1608.08940 (2016). + * + * Produces embeddings for elements in sequences using a hash-based approach to avoid storing a + * vocabulary. Uses MurmurHash3 for hashing elements to embedding indices and signs. + * + * Output DataFrame has columns "id" (element identifier, same type as sequence elements) and + * "vector" (dense vector of doubles, summed across all occurrences). + * + * Tradeoffs: Higher numPartitions reduces local state and memory per partition but increases + * aggregation and merging overhead across partitions. Larger embeddingsDim provides richer + * representations but consumes more memory. Seeds control hashing for reproducibility. + */ +class Hash2Vec extends Serializable { + private def decayGaussian(d: Int, sigma: Double): Double = { + math.exp(-(d * d) / (sigma * sigma)) + } + private val possibleDecayFunctions: Seq[String] = Seq("gaussian", "constant") + + private var contextSize: Int = 5 + private var numPartitions: Int = 5 + private var embeddingsDim: Int = 512 + private var sequenceCol: String = RandomWalkBase.rwColName + private var decayFunction: String = "gaussian" + private var gaussianSigma: Double = 1.0 + private var hashingSeed: Int = 42 + private var signHashingSeed: Int = 18 + private var doL2Norm: Boolean = true + private var safeL2NormAsChannel: Boolean = true + private var maxVectorsPerPartition: Int = 100000 + + /** + * Sets whether final vectors are L2-normalized after aggregation across partitions. When + * normalization is enabled, each vector is scaled to unit length (L2 norm = 1). + * + * When `safeNorm` is true (default), the method adds an extra channel to the vector equal to + * `log(L2-norm + 1) / sqrt(dim)`. This preserves some information about the original magnitude + * while still making vectors comparable via cosine similarity. + * + * When `safeNorm` is false, normalizes without adding an extra channel, discarding magnitude + * entirely. + * + * This setting applies globally to all output vectors. + * + * @param doNorm + * If true, output vectors are normalized. + * @param safeNorm + * If true (and doNorm is true), retains magnitude information in an extra dimension. If + * false, performs standard L2-normalization. + * @return + * This Hash2Vec instance for method chaining. + */ + def setDoNormalization(doNorm: Boolean, safeNorm: Boolean): this.type = { + doL2Norm = doNorm + safeL2NormAsChannel = safeNorm + this + } + + /** + * Limits the maximum number of distinct element vectors that can be processed inside a single + * partition before flushing intermediate results to the iterator. + * + * Partition processing uses a paged matrix to store vectors. When the number of allocated + * vectors reaches this limit within a partition, the current batch of vectors is returned (as + * part of the iterator) and a new empty batch is started for the remaining elements. + * + * This prevents a single partition from consuming unbounded memory while processing very large + * vocabularies, at the cost of producing multiple iterator batches per partition. + * + * The default value is 100000. + * + * @param value + * Upper bound on distinct vectors processed per batch inside a partition. + * @return + * This Hash2Vec instance for method chaining. + */ + def setMaxVectorsPerPartition(value: Int): this.type = { + maxVectorsPerPartition = value + this + } + + /** + * Convenience overload for `setDoNormalization(doNorm, safeNorm)` that uses safe-mode (extra + * channel) by default. Equivalent to `setDoNormalization(value, true)`. + * + * @param value + * If true, output vectors are L2-normalized with safe (extra-channel) semantics. + * @return + * This Hash2Vec instance for method chaining. + */ + def setDoNormalization(value: Boolean): this.type = { + setDoNormalization(value, true) + this + } + + /** + * Sets the context window size around each element to consider during training. Larger values + * incorporate more distant elements but increase computation time. Default: 5. + */ + def setContextSize(value: Int): this.type = { + contextSize = value + this + } + + /** + * Sets the number of partitions for RDDs to parallelize computation. More partitions distribute + * workload and reduce memory per partition but complicate merging across partitions. Default: + * 5. + */ + def setNumPartitions(value: Int): this.type = { + numPartitions = value + this + } + + /** + * Sets the dimensionality of the dense embedding vectors. Larger dimensions allow richer + * representations but require more memory. Corresponds to the hash table size. Default: 512. + */ + def setEmbeddingsDim(value: Int): this.type = { + embeddingsDim = value + this + } + + /** + * Sets the column name containing sequences of elements (as arrays). Default: "random_walk". + */ + def setSequenceCol(value: String): this.type = { + sequenceCol = value + this + } + + /** + * Sets the decay function used to weight context elements by distance. Supported values: + * "gaussian", "constant". Default: "gaussian". + */ + def setDecayFunction(value: String): this.type = { + val sep = ", " + require( + possibleDecayFunctions.contains(value), + s"supported functions: ${possibleDecayFunctions.mkString(sep)}") + decayFunction = value + this + } + + /** + * Sets the sigma parameter for Gaussian decay weighting. Smaller values decay weights faster + * with distance. Default: 1.0. + */ + def setGaussianSigma(value: Double): this.type = { + gaussianSigma = value + this + } + + /** + * Sets the seed for hashing elements to embedding indices. Used for reproducibility of + * embeddings. Default: 42. + */ + def setHashingSeed(value: Int): this.type = { + hashingSeed = value + this + } + + /** + * Sets the seed for hashing elements to determine the sign of contributions. Used for + * reproducibility of embeddings. Default: 18. + */ + def setSignHashSeed(value: Int): this.type = { + signHashingSeed = value + this + } + + private def nonNegativeMod(x: Int, mod: Int): Int = { + val rawMod = x % mod + rawMod + (if (rawMod < 0) mod else 0) + } + + private var weightFunction: (Int) => Double = _ + + private def seededStringHashFunc(s: String, seed: Int, dim: Int): Int = { + var h = seed + var i = 0 + val len = s.length + while (i < len) { + h = MurmurHash3.mix(h, s.charAt(i).toInt) + i += 1 + } + nonNegativeMod(MurmurHash3.finalizeHash(h, len), dim) + } + + private def seededLongHashFunc(l: Long, seed: Int, dim: Int): Int = + nonNegativeMod(hashLong(l, seed), dim) + + private def normalize(vector: linalg.Vector, addChannel: Boolean): linalg.Vector = { + val blas = BLAS.getInstance() + val arr = vector.toArray + val norm = blas.dnrm2(arr.size, arr, 0, 1) + blas.dscal(arr.size, 1 / (norm + 1e-6), arr, 1) + if (addChannel) { + val scaledL2 = math.log(norm + 1) + val newChannel = scaledL2 / math.sqrt(vector.size.toDouble) + new linalg.DenseVector(arr :+ newChannel) + } else { + new linalg.DenseVector(arr) + } + } + + /** + * Runs the Hash2Vec algorithm on the input DataFrame containing sequences. The specified + * sequenceCol must contain arrays of elements (string or numeric). Produces a DataFrame with + * "id" (element ID, same type as elements) and "vector" (embedding vector, VectorType). + * Embeddings are summed across all partitions and occurrences. + */ + def run(rawData: DataFrame): DataFrame = { + val spark = rawData.sparkSession + require( + rawData.schema(sequenceCol).dataType.isInstanceOf[ArrayType], + "sequence should be array") + val elDataType = rawData.schema(sequenceCol).dataType.asInstanceOf[ArrayType].elementType + + val data = + if (elDataType.isInstanceOf[ByteType] || elDataType.isInstanceOf[ShortType] || elDataType + .isInstanceOf[IntegerType]) { + rawData.withColumn( + sequenceCol, + transform(col(sequenceCol), (f: Column) => f.cast(LongType))) + } else { + rawData + } + + weightFunction = decayFunction match { + case "gaussian" => (d: Int) => decayGaussian(d, gaussianSigma) + case "constant" => (_: Int) => 1.0 + case _ => throw new RuntimeException(s"unsupported decay functions $decayFunction") + } + + val (rowRDD, schema) = elDataType match { + case _: StringType => + ( + runTyped[String](data).map(f => Row(f._1, Vectors.dense(f._2))), + StructType(Seq(StructField("id", StringType), StructField("vector", VectorType)))) + case _: LongType => + ( + runTyped[Long](data).map(f => Row(f._1, Vectors.dense(f._2))), + StructType(Seq(StructField("id", LongType), StructField("vector", VectorType)))) + case _ => + throw new GraphFramesUnsupportedVertexTypeException( + "Hash2vec supports only string or numeric types of elements but got " + + elDataType.toString) + } + + val embeddings = spark + .createDataFrame(rowRDD, schema) + .groupBy("id") + .agg(Summarizer.sum(col("vector")).alias("vector")) + + val normalizer = (x: linalg.Vector) => normalize(x, safeL2NormAsChannel) + + if (doL2Norm) { + embeddings.withColumn("vector", udf(normalizer).apply(col("vector"))) + } else { + embeddings + } + } + + private def runTyped[T: ClassTag](data: DataFrame): RDD[(T, Array[Double])] = { + // we should put sequences starts from the same vertex + // to the same partition when possible + data + .withColumn("hash_id", pmod(hash(col(sequenceCol).getItem(0)), lit(numPartitions))) + .repartition(numPartitions, col("hash_id")) + .sortWithinPartitions(col("hash_id")) + .select(col(sequenceCol)) + .rdd + .map(_.getSeq[T](0)) + .mapPartitions { iter => + val elemType = implicitly[ClassTag[T]].runtimeClass + elemType match { Review Comment: This is the only real part that needs to be updated to make it generic and fix is mechanical. I can do so. The only reason why it is not extending ML classes was "it was just easier to do" but with migration to the Apache Spark we can make it a nice addition to the ML seq2vec models family for cases that word2vec cannot handle. hash2vec is the easiest model of the class "extreme scale embeddings", perfect case for anyone who needs the "good enough baseline" for graph problems, item-product problems and other similar where the vocab exceed the word2vec limits. ########## graphframes/src/main/scala/org/apache/spark/graphframes/propertygraph/PropertyGraphFrame.scala: ########## Review Comment: This one needs to be changed a little to implement the queries. See https://github.com/graphframes/graphframes/pull/849 for details and another issue is https://github.com/graphframes/graphframes/issues/721 The Spark' policy about breaking changes is even more strict than policy in GF, so either we need to include 829 or does not include this one at all into the initial migration. ########## graphframes/src/test/scala/org/apache/spark/graphframes/ldbc/TestLDBCCases.scala: ########## @@ -0,0 +1,258 @@ +/* + * 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.spark.graphframes.ldbc + +import java.io.File +import java.nio.file._ +import java.util.Properties + +import org.apache.spark.graphframes.GraphFrame +import org.apache.spark.graphframes.GraphFrameTestSparkContext +import org.apache.spark.graphframes.SparkFunSuite +import org.apache.spark.graphframes.examples.LDBCUtils +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.functions.abs +import org.apache.spark.sql.functions.col +import org.apache.spark.sql.functions.lit +import org.apache.spark.sql.functions.sum +import org.apache.spark.sql.types.DoubleType +import org.apache.spark.sql.types.IntegerType +import org.apache.spark.sql.types.LongType +import org.apache.spark.sql.types.StructField +import org.apache.spark.sql.types.StructType + +class TestLDBCCases extends SparkFunSuite with GraphFrameTestSparkContext { Review Comment: LDBC files are very small and distributed under the Apache 2.0... What do you think about adding them to "resources" instead of downloading on the fly (using `downloadLDBCIfNotExists`)? They are really [small](https://ldbcouncil.org/benchmarks/graphalytics/datasets/), like 1-2 KiB. Should not add a lot to the repo size <img width="1043" height="815" alt="image" src="https://github.com/user-attachments/assets/309c8b76-d25e-4bb2-bf60-b3a0bc6bcee8"; /> ########## graphframes/src/main/scala/org/apache/spark/graphframes/lib/RandomizedContraction.scala: ########## @@ -0,0 +1,289 @@ +/* + * 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.spark.graphframes.lib + +import java.io.IOException +import java.util.UUID + +import scala.collection.mutable.Stack +import scala.util.Random + +import org.apache.hadoop.fs.Path + +import org.apache.spark.graphframes.GraphFrame +import org.apache.spark.graphframes.GraphFrame.{DST, ID, LONG_DST, LONG_ID, LONG_SRC, SRC} +import org.apache.spark.graphframes.Logging +import org.apache.spark.sql.Column +import org.apache.spark.sql.DataFrame +import org.apache.spark.sql.catalyst.FunctionIdentifier +import org.apache.spark.sql.catalyst.expressions.Expression +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.graphframes.expressions.FiniteAXPlusB +import org.apache.spark.storage.StorageLevel + +/** + * Implementation of parallel connected components algorithm using randomized contraction, based + * on Boegeholz, Harald, Michael Brand, and Radu-Alexandru Todor. "In-database connected component + * analysis." 2020 IEEE 36th International Conference on Data Engineering (ICDE). IEEE, 2020. + * + * The algorithm contracts the graph iteratively using random linear functions, until no edges + * remain, then reconstructs the component identifiers. + */ +private[graphframes] object RandomizedContraction extends Logging with Serializable { + private val CHECKPOINT_NAME_PREFIX = "randomized-contraction" + + private def prepare(graph: GraphFrame): GraphFrame = { + val vertices = graph.indexedVertices + .select(col(LONG_ID).as(ID)) + + val edges = graph.indexedEdges + .select(col(LONG_SRC).as(SRC), col(LONG_DST).as(DST)) + val symmetricEdges = edges + .union(edges.select(col(DST).alias(SRC), col(SRC).alias(DST))) + .distinct() + GraphFrame(vertices, symmetricEdges) + + } + + def run( + inputGraph: GraphFrame, + useLabelsAsComponents: Boolean, + intermediateStorageLevel: StorageLevel, + useLocalCheckpoints: Boolean, + checkpointInterval: Int, + isGraphPrepared: Boolean): DataFrame = { + val spark = inputGraph.vertices.sparkSession + val sc = spark.sparkContext + val runId = UUID.randomUUID().toString.takeRight(8) + val logPrefix = s"[CC $runId]" + + val checkpointDir = sc.getCheckpointDir + .map { d => + new Path(d, s"$CHECKPOINT_NAME_PREFIX-$runId").toString + } + .getOrElse { + // Spark-Connect workaround + spark.conf.getOption("spark.checkpoint.dir") match { Review Comment: It was my decision, but right now I'm not 100% it was a good idea to write parquet-files directly to the `spark.checkpoint.dir`. I'm not sure we need to keep as is. Maybe a separate argument, like a `workDir`? WDYT? -- 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]
