Revert operation type changes

Author: TheovanKraay

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/AsyncItemWriter.scala

@@ -14,14 +14,6 @@ private  trait AsyncItemWriter {
     */
   def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode): Unit
 
-  /**
-    * Schedule a write to happen in async and return immediately with per-row operation type
-    * @param partitionKeyValue the partition key value
-    * @param objectNode the json object node
-    * @param operationType optional operation type (create, upsert, replace, delete) for this specific row
-    */
-  def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode, operationType: Option[String]): Unit
-
   /**
     * Wait for all remaining work
     * Throws if any of the work resulted in failure

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/BulkWriter.scala

@@ -646,12 +646,6 @@ private class BulkWriter
   }
 
   override def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode): Unit = {
-    scheduleWrite(partitionKeyValue, objectNode, None)
-  }
-
-  override def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode, operationType: Option[String]): Unit = {
-    // BulkWriter doesn't support per-row operation types - it uses global ItemWriteStrategy
-    // The operationType parameter is ignored here for interface compatibility
     Preconditions.checkState(!closed.get())
     throwIfCapturedExceptionExists()
 

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/CosmosWriterBase.scala

@@ -97,23 +97,6 @@ private abstract class CosmosWriterBase(
   override def write(internalRow: InternalRow): Unit = {
     val objectNode = cosmosRowConverter.fromInternalRowToObjectNode(internalRow, inputSchema)
 
-    // Extract operationType if column exists (for per-row operation support)
-    val operationType: Option[String] = if (inputSchema.fieldNames.contains("operationType")) {
-      val opTypeIndex = inputSchema.fieldIndex("operationType")
-      if (!internalRow.isNullAt(opTypeIndex)) {
-        Some(internalRow.getString(opTypeIndex))
-      } else {
-        None
-      }
-    } else {
-      None
-    }
-
-    // Remove operationType from objectNode if present (don't persist to Cosmos)
-    if (objectNode.has("operationType")) {
-      objectNode.remove("operationType")
-    }
-
     require(objectNode.has(CosmosConstants.Properties.Id) &&
       objectNode.get(CosmosConstants.Properties.Id).isTextual,
       s"${CosmosConstants.Properties.Id} is a mandatory field. " +
@@ -127,16 +110,7 @@ private abstract class CosmosWriterBase(
     }
 
     val partitionKeyValue = PartitionKeyHelper.getPartitionKeyPath(objectNode, partitionKeyDefinition)
-    
-    // Call the appropriate scheduleWrite method based on whether operationType is specified
-    operationType match {
-      case Some(opType) =>
-        // Per-row operation type specified - use 3-parameter method
-        writer.get.scheduleWrite(partitionKeyValue, objectNode, Some(opType))
-      case None =>
-        // No per-row operation type - use 2-parameter method (global strategy)
-        writer.get.scheduleWrite(partitionKeyValue, objectNode)
-    }
+    writer.get.scheduleWrite(partitionKeyValue, objectNode)
   }
 
   override def commit(): WriterCommitMessage = {

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/PointWriter.scala

@@ -76,10 +76,6 @@ private class PointWriter(container: CosmosAsyncContainer,
   }
 
   override def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode): Unit = {
-    scheduleWrite(partitionKeyValue, objectNode, None)
-  }
-
-  override def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode, operationType: Option[String]): Unit = {
     checkState(!closed.get())
 
     val etag = getETag(objectNode)

sdk/cosmos/azure-cosmos-spark_3/src/main/scala/com/azure/cosmos/spark/TransactionalBulkWriter.scala

@@ -411,18 +411,6 @@ private class TransactionalBulkWriter
     scheduleWriteInternal(partitionKeyValue, objectNode, operationContext)
   }
 
-  /**
-    * Per-row operation type is not supported for transactional batches.
-    * All operations in a transactional batch must be upserts (ItemOverwrite strategy).
-    * This method is implemented to satisfy the AsyncItemWriter interface but throws an exception.
-    */
-  override def scheduleWrite(partitionKeyValue: PartitionKey, objectNode: ObjectNode, operationType: Option[String]): Unit = {
-    throw new UnsupportedOperationException(
-      "Per-row operation types are not supported for transactional batches. " +
-      "All operations in a transactional batch must use ItemOverwrite (upsert) strategy. " +
-      "Use scheduleWrite(partitionKeyValue, objectNode) instead.")
-  }
-
   private def scheduleWriteInternal(partitionKeyValue: PartitionKey,
                                     objectNode: ObjectNode,
                                     operationContext: OperationContext): Unit = {