You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "aglinxinyuan (via GitHub)" <gi...@apache.org> on 2026/05/04 06:59:44 UTC

[PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

aglinxinyuan opened a new pull request, #4912:
URL: https://github.com/apache/texera/pull/4912

   ## Summary
   - New `ProgressiveUtilsSpec` (10 tests) pinning the insertion / retraction flag contract: `insertRetractFlagAttr` shape, `addInsertionFlag` / `addRetractionFlag` value + double-flag-assertion, `isInsertion` defaulting + flag-reading, `getTupleFlagAndValue` stripping behavior including unflagged-passthrough.
   
   Closes #4911
   
   ## Test plan
   - [x] `sbt 'WorkflowOperator/Test/testOnly org.apache.texera.amber.operator.sink.ProgressiveUtilsSpec'` — all 10 cases pass.
   - [x] `sbt 'WorkflowOperator/Test/scalafmtCheck'` — clean.
   
   🤖 Generated with [Claude Code](https://claude.com/claude-code)


-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180201166


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    // Full schema equality (names + types + order) — name-only would let a
+    // type drift on the payload columns slip through.
+    assert(stripped.getSchema == baseSchema)
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getSchema == baseSchema)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass an equivalent schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns an equivalent Schema — same attributes
+    // in the same order (filterNot removes nothing). Note that
+    // Schema.getPartialSchema constructs a new instance every time, so this
+    // is structural equality, not reference identity.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema == raw.getSchema)
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Use value-based equality (the Tuple contract elsewhere uses
+    // `sameElements` for Array[Byte]); requiring the *same* array instance
+    // would over-constrain the flag/strip path against future copy-on-write
+    // changes.
+    assert(value.asInstanceOf[Array[Byte]].sameElements(bytes))
+  }
+
+  it should "preserve null payload values across all types" in {
+    Seq(
+      new Attribute("v_int", AttributeType.INTEGER),
+      new Attribute("v_str", AttributeType.STRING),
+      new Attribute("v_dbl", AttributeType.DOUBLE),
+      new Attribute("v_bool", AttributeType.BOOLEAN),
+      new Attribute("v_ts", AttributeType.TIMESTAMP)

Review Comment:
   The test name says it preserves null payload values "across all types", but the list only covers INTEGER/STRING/DOUBLE/BOOLEAN/TIMESTAMP and omits other defined AttributeTypes (e.g., LONG, BINARY, LARGE_BINARY, ANY). Either broaden the tested attributes to match the claim or rename the test to avoid implying full coverage.
   



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #4912:
URL: https://github.com/apache/texera/pull/4912#issuecomment-4368902361

   ## [Codecov](https://app.codecov.io/gh/apache/texera/pull/4912?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :white_check_mark: All modified and coverable lines are covered by tests.
   :white_check_mark: Project coverage is 42.03%. Comparing base ([`06831ff`](https://app.codecov.io/gh/apache/texera/commit/06831ff568edba9649990e8726abdbd305486868?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8f851c3`](https://app.codecov.io/gh/apache/texera/commit/8f851c37783f77d8f5bf084885648a6805e8ae97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 1 commits behind head on main.
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@            Coverage Diff            @@
   ##               main    #4912   +/-   ##
   =========================================
     Coverage     42.03%   42.03%           
     Complexity     2155     2155           
   =========================================
     Files           980      980           
     Lines         36292    36292           
     Branches       3783     3783           
   =========================================
     Hits          15256    15256           
     Misses        20112    20112           
     Partials        924      924           
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/texera/pull/4912/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [config-service](https://app.codecov.io/gh/apache/texera/pull/4912/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/texera/pull/4912?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179924431


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+    // Don't pin the exact message; locking in the AssertionError is enough.
+    assert(ex != null)
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {

Review Comment:
   Looked for it — there is no public resolution API in the current codebase. `ProgressiveUtils` only exposes producers (`addInsertionFlag` / `addRetractionFlag`) and readers (`isInsertion` / `getTupleFlagAndValue`); there is no shared utility (e.g., a deduplicating fold over a flagged stream that produces a materialized `Set[Tuple]`). I checked usages of `insertRetractFlagAttr`, `__internal_is_insertion`, and `getTupleFlagAndValue` — only the file itself references them, no downstream consumer applies retractions to undo prior insertions.
   
   Are you thinking of an API that used to exist and has since been removed, or one that *should* exist? Happy to either (a) add a `materialize(stream): Set[Tuple]` helper here and test it as part of this PR, or (b) leave that as a follow-up issue. Let me know which you'd prefer.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180093386


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.

Review Comment:
   Done in d3c1ac8273 — reworded the inline comment from `"returns the same schema"` to `"returns an equivalent Schema — same attributes in the same order"` and explicitly noted that `Schema.getPartialSchema` constructs a fresh instance each call. Also strengthened the assertion from a name-list comparison to a full `stripped.getSchema == raw.getSchema` structural-equality check.



##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Use value-based equality (the Tuple contract elsewhere uses
+    // `sameElements` for Array[Byte]); requiring the *same* array instance
+    // would over-constrain the flag/strip path against future copy-on-write
+    // changes.
+    assert(value.asInstanceOf[Array[Byte]].sameElements(bytes))

Review Comment:
   This was already addressed earlier in this PR — c1783d1751 switched the BINARY assertion from `eq` (reference) to `sameElements` (value-based, matching `Tuple.equals`'s contract). Looks like this thread was opened against the pre-fix version. The current spec on line ~199 reads `assert(value.asInstanceOf[Array[Byte]].sameElements(bytes))`.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179885502


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )

Review Comment:
   let's test more types' incremental behavior. does it support float, double, boolean, timestamp? oh and binary?



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180064618


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+    // Don't pin the exact message; locking in the AssertionError is enough.
+    assert(ex != null)
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {

Review Comment:
   Made the call: opened #4920 to track adding a `materialize(stream): Set[Tuple]` (or similar) API as a follow-up. Keeping this PR scoped to testing the existing `ProgressiveUtils` surface — adding a new public API plus its tests would expand the change beyond what the PR title and existing reviewers signed up for. The follow-up issue captures the suggested shape and open questions (ordering preserved? handling of out-of-order retractions?) so an implementer can pick it up cleanly.
   
   Resolving this thread; flag back if you'd rather have it folded into this PR after all.



##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Tuple stores the same array reference through the round-trip (no copy
+    // semantics in the flag/strip path), so reference equality is the
+    // observable contract here.
+    assert(value.asInstanceOf[Array[Byte]] eq bytes)

Review Comment:
   Done in c1783d1751 — switched the BINARY assertion from reference equality (`eq`) to `sameElements`. The Tuple contract already uses `sameElements` for Array[Byte], and reference equality would have over-constrained the flag/strip path against any future copy-on-write 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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180070291


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Use value-based equality (the Tuple contract elsewhere uses
+    // `sameElements` for Array[Byte]); requiring the *same* array instance
+    // would over-constrain the flag/strip path against future copy-on-write
+    // changes.
+    assert(value.asInstanceOf[Array[Byte]].sameElements(bytes))

Review Comment:
   This test asserts reference equality (`eq`) for BINARY payloads and documents it as an “observable contract”. That’s likely over-constraining: elsewhere in the codebase `Tuple.equals` treats `Array[Byte]` with value semantics (`sameElements`). Suggest asserting byte content equality (e.g., `sameElements`) rather than identity so the test doesn’t fail if the implementation later copies arrays.



##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.

Review Comment:
   The inline comment claims `getPartialSchema` “returns the same schema” for an unflagged tuple, but `Schema.getPartialSchema(...)` always constructs a new `Schema` instance (even if it contains the same attributes). Consider rewording this to “an equivalent schema” and/or assert schema equality explicitly (e.g., `stripped.getSchema == raw.getSchema`) if that’s the intended contract.
   



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180166737


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")

Review Comment:
   Done in 118e837957 — switched the `getTupleFlagAndValue` assertions from `getAttributeNames == List(...)` (names + order only) to a full `stripped.getSchema == baseSchema` equality check, so any payload-column type drift would also fail the test.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179854691


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)

Review Comment:
   The “already has the flag column” case is only exercised by calling addRetractionFlag on a flagged tuple. Since both addInsertionFlag and addRetractionFlag contain the same guard (and either could be called on already-flagged data), add an additional test that addInsertionFlag also throws when the input tuple already carries the flag attribute.



##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+    // Don't pin the exact message; locking in the AssertionError is enough.
+    assert(ex != null)

Review Comment:
   intercept[...] already fails the test if no exception is thrown and returns the caught exception otherwise, so `assert(ex != null)` is redundant here and can be removed to keep the test focused on the behavior under test.
   



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180121190


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")

Review Comment:
   In the getTupleFlagAndValue tests for flagged tuples, `stripped.getSchema.getAttributeNames == List("id", "name")` only asserts the names/order, not that the stripped schema’s attribute *types* match the expected base schema. To fully pin the contract described in the PR, consider asserting `stripped.getSchema == baseSchema` (or otherwise verifying attribute types) for both insertion- and retraction-flagged cases.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179923005


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )

Review Comment:
   Done in 6eb484791d — added explicit flag → strip round-trip tests for INTEGER, LONG, DOUBLE, BOOLEAN (specifically distinct from the flag column's own BOOLEAN type, to verify the implementation selects by name not type), TIMESTAMP, BINARY, plus null payloads across types. The flag/strip path is type-agnostic, but pinning these catches a payload-eating regression that would otherwise surface only in a downstream sink test.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3180301071


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    // Full schema equality (names + types + order) — name-only would let a
+    // type drift on the payload columns slip through.
+    assert(stripped.getSchema == baseSchema)
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getSchema == baseSchema)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass an equivalent schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns an equivalent Schema — same attributes
+    // in the same order (filterNot removes nothing). Note that
+    // Schema.getPartialSchema constructs a new instance every time, so this
+    // is structural equality, not reference identity.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema == raw.getSchema)
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Use value-based equality (the Tuple contract elsewhere uses
+    // `sameElements` for Array[Byte]); requiring the *same* array instance
+    // would over-constrain the flag/strip path against future copy-on-write
+    // changes.
+    assert(value.asInstanceOf[Array[Byte]].sameElements(bytes))
+  }
+
+  it should "preserve null payload values across all types" in {
+    Seq(
+      new Attribute("v_int", AttributeType.INTEGER),
+      new Attribute("v_str", AttributeType.STRING),
+      new Attribute("v_dbl", AttributeType.DOUBLE),
+      new Attribute("v_bool", AttributeType.BOOLEAN),
+      new Attribute("v_ts", AttributeType.TIMESTAMP)

Review Comment:
   Done in the latest commit — replaced the hand-listed 5 types with `AttributeType.values.foreach`, so every enum member (STRING/INTEGER/LONG/DOUBLE/BOOLEAN/TIMESTAMP/BINARY/LARGE_BINARY/ANY) is covered. A future addition to the enum is automatically tested.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179889927


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+    // Don't pin the exact message; locking in the AssertionError is enough.
+    assert(ex != null)
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {

Review Comment:
   not sure if this is the api, but there should be an API to resolve those flags to get a final value. have we tested it?



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179902025


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)

Review Comment:
   Done in the latest commit — added a symmetric test that `addInsertionFlag` also throws `AssertionError` when called on an already-flagged tuple, paralleling the existing `addRetractionFlag` case.



##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if the input tuple already has the flag column" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    val ex = intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+    // Don't pin the exact message; locking in the AssertionError is enough.
+    assert(ex != null)

Review Comment:
   Done in the latest commit — dropped the redundant `assert(ex != null)`. `intercept[...]` already fails the test if no exception is thrown, so the null check was dead code.



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4912:
URL: https://github.com/apache/texera/pull/4912#discussion_r3179995431


##########
common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sink/ProgressiveUtilsSpec.scala:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.texera.amber.operator.sink
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple}
+import org.scalatest.flatspec.AnyFlatSpec
+
+class ProgressiveUtilsSpec extends AnyFlatSpec {
+
+  // --- helpers ---------------------------------------------------------------
+
+  private val baseSchema: Schema = new Schema(
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  // outputSchema = flag column prepended to baseSchema
+  private val outputSchema: Schema = new Schema(
+    ProgressiveUtils.insertRetractFlagAttr,
+    new Attribute("id", AttributeType.INTEGER),
+    new Attribute("name", AttributeType.STRING)
+  )
+
+  private def baseTuple(id: Int, name: String): Tuple =
+    Tuple
+      .builder(baseSchema)
+      .add(new Attribute("id", AttributeType.INTEGER), Int.box(id))
+      .add(new Attribute("name", AttributeType.STRING), name)
+      .build()
+
+  // --- insertRetractFlagAttr -------------------------------------------------
+
+  "ProgressiveUtils.insertRetractFlagAttr" should "be a BOOLEAN attribute named __internal_is_insertion" in {
+    val attr = ProgressiveUtils.insertRetractFlagAttr
+    assert(attr.getName == "__internal_is_insertion")
+    assert(attr.getType == AttributeType.BOOLEAN)
+  }
+
+  // --- addInsertionFlag / addRetractionFlag ----------------------------------
+
+  "ProgressiveUtils.addInsertionFlag" should "prepend the flag column with value true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    assert(flagged.getSchema == outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == true)
+    assert(flagged.getField[Integer]("id") == 1)
+    assert(flagged.getField[String]("name") == "alice")
+  }
+
+  "ProgressiveUtils.addRetractionFlag" should "prepend the flag column with value false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    assert(flagged.getField[Boolean](ProgressiveUtils.insertRetractFlagAttr.getName) == false)
+    assert(flagged.getField[Integer]("id") == 2)
+    assert(flagged.getField[String]("name") == "bob")
+  }
+
+  it should "fail an assertion if addRetractionFlag is called on an already-flagged tuple" in {
+    val alreadyFlagged = ProgressiveUtils.addInsertionFlag(baseTuple(3, "x"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addRetractionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  it should "fail an assertion if addInsertionFlag is called on an already-flagged tuple" in {
+    // Symmetric guard: both addInsertionFlag and addRetractionFlag carry the
+    // same `assert(!containsAttribute(flagAttr))` precondition, and either
+    // one may be called on already-flagged data, so each path should fail.
+    val alreadyFlagged = ProgressiveUtils.addRetractionFlag(baseTuple(4, "y"), outputSchema)
+    intercept[AssertionError] {
+      ProgressiveUtils.addInsertionFlag(alreadyFlagged, outputSchema)
+    }
+  }
+
+  // --- isInsertion -----------------------------------------------------------
+
+  "ProgressiveUtils.isInsertion" should "return true for an unflagged tuple" in {
+    // Tuples without the flag column default to insertion (the unflagged
+    // default in the engine is "+").
+    assert(ProgressiveUtils.isInsertion(baseTuple(1, "x")))
+  }
+
+  it should "return true when the flag column is present and true" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "x"), outputSchema)
+    assert(ProgressiveUtils.isInsertion(flagged))
+  }
+
+  it should "return false when the flag column is present and false" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(1, "x"), outputSchema)
+    assert(!ProgressiveUtils.isInsertion(flagged))
+  }
+
+  // --- getTupleFlagAndValue --------------------------------------------------
+
+  "ProgressiveUtils.getTupleFlagAndValue" should "split an insertion-flagged tuple into (true, base tuple)" in {
+    val flagged = ProgressiveUtils.addInsertionFlag(baseTuple(1, "alice"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 1)
+    assert(stripped.getField[String]("name") == "alice")
+  }
+
+  it should "split a retraction-flagged tuple into (false, base tuple)" in {
+    val flagged = ProgressiveUtils.addRetractionFlag(baseTuple(2, "bob"), outputSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    assert(!flag)
+    assert(stripped.getField[Integer]("id") == 2)
+    assert(stripped.getField[String]("name") == "bob")
+  }
+
+  it should "treat an unflagged tuple as insertion and pass the original schema through unchanged" in {
+    // For a tuple that doesn't carry the flag column, isInsertion returns
+    // true and getPartialSchema returns the same schema (filterNot removes
+    // nothing). The values must round-trip intact.
+    val raw = baseTuple(3, "carol")
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(raw)
+    assert(flag)
+    assert(stripped.getSchema.getAttributeNames == List("id", "name"))
+    assert(stripped.getField[Integer]("id") == 3)
+    assert(stripped.getField[String]("name") == "carol")
+  }
+
+  // --- typed payload round-trips --------------------------------------------
+  // Nothing in `addInsertionFlag` / `getTupleFlagAndValue` is type-specific —
+  // they only care about the BOOLEAN flag column they prepend / strip — but
+  // it is worth pinning that arbitrary AttributeType payload columns survive
+  // the flag → strip → unflag round-trip across the engine's value types.
+
+  private def flagRoundTrip(payloadAttr: Attribute, payloadValue: AnyRef): (Boolean, AnyRef) = {
+    val payloadSchema = new Schema(payloadAttr)
+    val flaggedSchema = new Schema(ProgressiveUtils.insertRetractFlagAttr, payloadAttr)
+    val raw = Tuple.builder(payloadSchema).add(payloadAttr, payloadValue).build()
+    val flagged = ProgressiveUtils.addInsertionFlag(raw, flaggedSchema)
+    val (flag, stripped) = ProgressiveUtils.getTupleFlagAndValue(flagged)
+    (flag, stripped.getField[AnyRef](payloadAttr.getName))
+  }
+
+  "Flag round-trip" should "preserve INTEGER payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.INTEGER), Int.box(42))
+    assert(flag)
+    assert(value == Int.box(42))
+  }
+
+  it should "preserve LONG payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.LONG), Long.box(9876543210L))
+    assert(flag)
+    assert(value == Long.box(9876543210L))
+  }
+
+  it should "preserve DOUBLE payload values" in {
+    val (flag, value) =
+      flagRoundTrip(new Attribute("v", AttributeType.DOUBLE), Double.box(3.14159))
+    assert(flag)
+    assert(value == Double.box(3.14159))
+  }
+
+  it should "preserve BOOLEAN payload values (distinct from the flag column)" in {
+    // The flag column is also BOOLEAN; this verifies the implementation
+    // selects the correct attribute by name, not by type.
+    val (flag, value) =
+      flagRoundTrip(new Attribute("active", AttributeType.BOOLEAN), Boolean.box(false))
+    assert(flag, "outer flag must still be insertion")
+    assert(value == Boolean.box(false), "inner BOOLEAN payload must be preserved")
+  }
+
+  it should "preserve TIMESTAMP payload values" in {
+    val ts = new java.sql.Timestamp(1_700_000_000_000L)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("ts", AttributeType.TIMESTAMP), ts)
+    assert(flag)
+    assert(value == ts)
+  }
+
+  it should "preserve BINARY payload values" in {
+    val bytes = Array[Byte](0, 1, 2, 3, -1)
+    val (flag, value) =
+      flagRoundTrip(new Attribute("blob", AttributeType.BINARY), bytes)
+    assert(flag)
+    // Tuple stores the same array reference through the round-trip (no copy
+    // semantics in the flag/strip path), so reference equality is the
+    // observable contract here.
+    assert(value.asInstanceOf[Array[Byte]] eq bytes)

Review Comment:
   The BINARY payload test asserts reference equality (`eq`) for the Array[Byte]. This over-constrains the behavior (and would fail if Tuple/ProgressiveUtils later copies arrays for immutability) while the tuple contract elsewhere is value-based (Tuple.equals uses `sameElements` for Array[Byte]). Consider asserting the byte contents instead (e.g., `sameElements`) rather than requiring the same array instance.
   



-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(workflow-operator): add unit test coverage for ProgressiveUtils [texera]

Posted by "aglinxinyuan (via GitHub)" <gi...@apache.org>.
aglinxinyuan merged PR #4912:
URL: https://github.com/apache/texera/pull/4912


-- 
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: notifications-unsubscribe@texera.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org