diff --git a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/LightGBMBase.scala b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/LightGBMBase.scala index 287aab262c4..cabd4604e5e 100644 --- a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/LightGBMBase.scala +++ b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/LightGBMBase.scala @@ -199,7 +199,7 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] private def getSlotNamesWithMetadata(featuresSchema: StructField): Option[Array[String]] = { if (getSlotNames.nonEmpty) { - Some(getSlotNames) + Some(ensureUniqueFeatureNames(getSlotNames)) } else { AttributeGroup.fromStructField(featuresSchema).attributes.flatMap(attributes => if (attributes.isEmpty) { @@ -208,28 +208,82 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] val colNames = attributes.indices.map(_.toString).toArray attributes.foreach(attr => attr.index.foreach(index => colNames(index) = attr.name.getOrElse(index.toString))) - Some(colNames) + // Ensure unique feature names to avoid LightGBM error: + // "Feature (Column_) appears more than one time" + // This can occur in Spark 3.5+ due to changes in AttributeGroup metadata handling + Some(ensureUniqueFeatureNames(colNames)) } ) } } - private def validateSlotNames(featuresSchema: StructField): Unit = { - val metadata = AttributeGroup.fromStructField(featuresSchema) - if (metadata.attributes.isDefined) { - val slotNamesOpt = getSlotNamesWithMetadata(featuresSchema) - val pattern = new Regex("[\",:\\[\\]{}]") - slotNamesOpt.foreach(slotNames => { - val badSlotNames = slotNames.flatMap(slotName => - if (pattern.findFirstIn(slotName).isEmpty) None else Option(slotName)) - if (!badSlotNames.isEmpty) { - throw new IllegalArgumentException( - s"Invalid slot names detected in features column: ${badSlotNames.mkString(",")}" + - " \n Special characters \" , : \\ [ ] { } will cause unexpected behavior in LGBM unless changed." + - " This error can be fixed by renaming the problematic columns prior to vector assembly.") - } - }) + /** + * Maps a feature name to the form LightGBM compares internally. LightGBM replaces every space + * with an underscore before checking for duplicates, so "a b" and "a_b" are the same feature + * as far as the native library is concerned even though they differ in Scala. + */ + private def normalizeFeatureName(name: String): String = name.replace(' ', '_') + + /** + * Ensures all feature names are unique by appending a numeric suffix to repeated names. + * LightGBM rejects a Dataset whose feature names repeat, failing the native + * LGBM_DatasetSetFeatureNames call with "Feature (X) appears more than one time", and Spark + * can surface repeated names through AttributeGroup metadata on the features column. + * + * Uniqueness is decided on the normalized form (see normalizeFeatureName), because that is + * what LightGBM compares. The original names are what get emitted, so this only affects which + * names are considered to collide. + * + * Every original name is reserved up front, so a generated name can never collide with an + * original that appears later in the array. Generated names are reserved as they are handed + * out, so they cannot collide with each other either. Order is preserved, which matters + * because feature names are positional in LightGBM. + * + * @param names The array of feature names that may contain duplicates. + * @return An array of unique feature names, in the original order. + */ + private def ensureUniqueFeatureNames(names: Array[String]): Array[String] = { + val reserved = scala.collection.mutable.HashSet[String](names.map(normalizeFeatureName): _*) + val emitted = scala.collection.mutable.HashSet[String]() + val renamed = scala.collection.mutable.ArrayBuffer[String]() + + val uniqueNames = names.map { name => + if (emitted.add(normalizeFeatureName(name))) { + name + } else { + // Terminates because only finitely many names are reserved. + val uniqueName = Iterator.from(1) + .map(suffix => s"${name}_$suffix") + .find(candidate => !reserved.contains(normalizeFeatureName(candidate))) + .get + reserved.add(normalizeFeatureName(uniqueName)) + emitted.add(normalizeFeatureName(uniqueName)) + renamed += name + uniqueName + } + } + + if (renamed.nonEmpty) { + log.warn(s"Duplicate feature names detected and renamed: ${renamed.distinct.mkString(", ")}. " + + "Set the 'slotNames' parameter explicitly to control feature naming.") } + + uniqueNames + } + + private def validateSlotNames(featuresSchema: StructField): Unit = { + val slotNamesOpt = getSlotNamesWithMetadata(featuresSchema) + val pattern = new Regex("[\",:\\[\\]{}]") + slotNamesOpt.foreach(slotNames => { + val badSlotNames = slotNames.flatMap(slotName => + if (pattern.findFirstIn(slotName).isEmpty) None else Option(slotName)) + if (!badSlotNames.isEmpty) { + throw new IllegalArgumentException( + s"Invalid slot names detected in features column: ${badSlotNames.mkString(",")}" + + " \n Special characters \" , : [ ] { } will cause unexpected behavior in LGBM unless changed." + + " This error can be fixed by renaming the problematic columns prior to vector assembly.") + } + }) } /** @@ -414,6 +468,10 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] val (numCols, numInitScoreClasses) = calculateColumnStatistics(preprocessedDF, measures) val featuresSchema = dataset.schema(getFeaturesCol) + // Validate before any native LightGBM call that consumes feature names (e.g. reference + // Dataset creation below), so an invalid name surfaces as an actionable error naming the + // offending columns rather than an opaque native failure. + validateSlotNames(featuresSchema) val generalTrainParams: BaseTrainParams = getTrainParams(numTasks, featuresSchema, numTasksPerExecutor) val trainParams = addCustomTrainParams(generalTrainParams, dataset) log.info(s"LightGBM batch $batchIndex of $batchCount, parameters: ${trainParams.toString()}") @@ -422,7 +480,7 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] val (serializedReferenceDataset: Option[Array[Byte]], partitionCounts: Option[Array[Long]]) = if (isStreamingMode) { val (referenceDataset, partitionCounts) = - calculateRowStatistics(trainingData, trainParams, numCols, measures) + calculateRowStatistics(trainingData, trainParams, numCols, featuresSchema, measures) // Save the reference Dataset so it's available to client and other batches if (getReferenceDataset.isEmpty) { @@ -432,7 +490,6 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] (Some(referenceDataset), Some(partitionCounts)) } else (None, None) - validateSlotNames(featuresSchema) executeTraining(preprocessedDF, validationData, serializedReferenceDataset, @@ -503,12 +560,14 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] * @param dataframe The dataset to train on. * @param trainingParams The training parameters. * @param numCols The number of feature columns. + * @param featuresSchema The schema of the features column. * @param measures Instrumentation measures. * @return The serialized Dataset reference and an array of partition counts. */ private def calculateRowStatistics(dataframe: DataFrame, trainingParams: BaseTrainParams, numCols: Int, + featuresSchema: StructField, measures: InstrumentationMeasures): (Array[Byte], Array[Long]) = { measures.markRowStatisticsStart() @@ -523,6 +582,9 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] trainingParams.generalParams.categoricalFeatures, trainingParams.executionParams.numThreads) + // Get feature names to set on the reference dataset (ensures unique names for Spark 3.5+) + val featureNames = getSlotNamesWithMetadata(featuresSchema) + // Either get a reference dataset (as bytes) from params, or calculate it val precalculatedDataset = getReferenceDataset val serializedReference = if (precalculatedDataset.nonEmpty) { @@ -541,6 +603,7 @@ trait LightGBMBase[TrainedModel <: Model[TrainedModel] with LightGBMModelParams] totalNumRows, numCols, collectedSampleData, + featureNames, measures, log) } diff --git a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/LightGBMDataset.scala b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/LightGBMDataset.scala index b841f2954c4..14b96de87a1 100644 --- a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/LightGBMDataset.scala +++ b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/LightGBMDataset.scala @@ -7,6 +7,7 @@ import com.microsoft.azure.synapse.ml.lightgbm.LightGBMUtils import com.microsoft.azure.synapse.ml.lightgbm.dataset.DatasetUtils.countCardinality import com.microsoft.lightgbm.SwigPtrWrapper import com.microsoft.ml.lightgbm._ +import org.slf4j.{Logger, LoggerFactory} import scala.reflect.ClassTag @@ -179,8 +180,17 @@ class LightGBMDataset(val datasetPtr: SWIGTYPE_p_void) extends AutoCloseable { // Add in slot names if they exist featureNamesOpt.foreach { featureNamesArray => if (featureNamesArray.nonEmpty) { - LightGBMUtils.validate(lightgbmlib.LGBM_DatasetSetFeatureNames(datasetPtr, featureNamesArray, numCols), - "Dataset set feature names") + // LGBM_DatasetSetFeatureNames reads numCols entries from the array, so a shorter array + // is an out-of-bounds native read. slotNames is user-supplied and unvalidated, so guard + // every dataset-naming path here rather than at individual call sites. LightGBM falls + // back to its own generated names when naming is skipped. + if (featureNamesArray.length != numCols) { + LightGBMDataset.Log.warn(s"Skipping feature names: got ${featureNamesArray.length} names " + + s"for $numCols feature columns.") + } else { + LightGBMUtils.validate(lightgbmlib.LGBM_DatasetSetFeatureNames(datasetPtr, featureNamesArray, numCols), + "Dataset set feature names") + } } } this @@ -191,3 +201,7 @@ class LightGBMDataset(val datasetPtr: SWIGTYPE_p_void) extends AutoCloseable { LightGBMUtils.validate(lightgbmlib.LGBM_DatasetFree(datasetPtr), "Free Dataset") } } + +object LightGBMDataset { + private val Log: Logger = LoggerFactory.getLogger(classOf[LightGBMDataset]) +} diff --git a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/ReferenceDatasetUtils.scala b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/ReferenceDatasetUtils.scala index 63743835f93..80b2e1585bf 100644 --- a/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/ReferenceDatasetUtils.scala +++ b/lightgbm/src/main/scala/com/microsoft/azure/synapse/ml/lightgbm/dataset/ReferenceDatasetUtils.scala @@ -16,25 +16,42 @@ object ReferenceDatasetUtils { numRows: Long, numCols: Int, sampledRowData: Array[Row], + featureNames: Option[Array[String]], measures: InstrumentationMeasures, log: Logger): Array[Byte] = { log.info(s"Creating reference training dataset with ${sampledRowData.length} samples and config: $datasetParams") - // Pre-create allocated native pointers so it's easy to clean them up - val datasetVoidPtr = lightgbmlib.voidpp_handle() val lenPtr = lightgbmlib.new_intp() val bufferHandlePtr = lightgbmlib.voidpp_handle() - val sampledData = SampledData(sampledRowData.length, numCols) + try { - // create properly formatted sampled data measures.markSamplingStart() sampledRowData.zipWithIndex.foreach({case (row, index) => sampledData.pushRow(row, index, featuresCol)}) measures.markSamplingStop() - // Create dataset from samples - // 1. Generate the dataset for features - val datasetVoidPtr = lightgbmlib.voidpp_handle() + val datasetHandle = createDatasetFromSamples(sampledData, numCols, numRows, datasetParams) + try { + setFeatureNamesIfProvided(datasetHandle, featureNames, numCols, log) + serializeReference(datasetHandle, bufferHandlePtr, lenPtr, log) + } finally { + // Free unconditionally so a failure while naming or serializing cannot leak the native + // Dataset. Deliberately not validated: throwing here would mask the original failure. + lightgbmlib.LGBM_DatasetFree(datasetHandle) + } + } finally { + sampledData.delete() + lightgbmlib.delete_voidpp(bufferHandlePtr) + lightgbmlib.delete_intp(lenPtr) + } + } + + private def createDatasetFromSamples(sampledData: SampledData, + numCols: Int, + numRows: Long, + datasetParams: String): SWIGTYPE_p_void = { + val datasetVoidPtr = lightgbmlib.voidpp_handle() + try { LightGBMUtils.validate(lightgbmlib.LGBM_DatasetCreateFromSampledColumn( sampledData.getSampleData, sampledData.getSampleIndices, @@ -45,31 +62,43 @@ object ReferenceDatasetUtils { numRows, datasetParams, datasetVoidPtr), "Dataset create from samples") - - - // 2. Serialize the raw dataset to a native buffer - val datasetHandle: SWIGTYPE_p_void = lightgbmlib.voidpp_value(datasetVoidPtr) - LightGBMUtils.validate(lightgbmlib.LGBM_DatasetSerializeReferenceToBinary( - datasetHandle, - bufferHandlePtr, - lenPtr), "Serialize ref") - val bufferLen: Int = lightgbmlib.intp_value(lenPtr) - log.info(s"Created serialized reference dataset of length $bufferLen") - - // The dataset is now serialized to a buffer, so we don't need original - LightGBMUtils.validate(lightgbmlib.LGBM_DatasetFree(datasetHandle), "Free Dataset") - - // This will also free the buffer - toByteArray(bufferHandlePtr, bufferLen) - } - finally { - sampledData.delete() + lightgbmlib.voidpp_value(datasetVoidPtr) + } finally { + // Frees the void** container only. The Dataset it points at is freed by LGBM_DatasetFree. lightgbmlib.delete_voidpp(datasetVoidPtr) - lightgbmlib.delete_voidpp(bufferHandlePtr) - lightgbmlib.delete_intp(lenPtr) } } + private def setFeatureNamesIfProvided(datasetHandle: SWIGTYPE_p_void, + featureNames: Option[Array[String]], + numCols: Int, + log: Logger): Unit = { + featureNames.foreach { names => + if (names.length != numCols) { + // LGBM_DatasetSetFeatureNames reads numCols entries from the array, so a shorter array + // would be an out-of-bounds native read. Skip naming rather than risk it; LightGBM then + // falls back to its own generated names, which is the behavior prior to this feature. + log.warn(s"Skipping feature names on reference dataset: got ${names.length} names " + + s"for $numCols feature columns.") + } else if (names.nonEmpty) { + log.info(s"Setting ${names.length} feature names on reference dataset") + LightGBMUtils.validate(lightgbmlib.LGBM_DatasetSetFeatureNames(datasetHandle, names, numCols), + "Dataset set feature names") + } + } + } + + private def serializeReference(datasetHandle: SWIGTYPE_p_void, + bufferHandlePtr: SWIGTYPE_p_p_void, + lenPtr: SWIGTYPE_p_int, + log: Logger): Array[Byte] = { + LightGBMUtils.validate(lightgbmlib.LGBM_DatasetSerializeReferenceToBinary( + datasetHandle, bufferHandlePtr, lenPtr), "Serialize ref") + val bufferLen: Int = lightgbmlib.intp_value(lenPtr) + log.info(s"Created serialized reference dataset of length $bufferLen") + toByteArray(bufferHandlePtr, bufferLen) + } + def getInitializedReferenceDataset(ctx: PartitionTaskContext): LightGBMDataset = { // The definition is broadcast from Spark, so retrieve it val serializedDataset: Array[Byte] = ctx.trainingCtx.serializedReferenceDataset.get diff --git a/lightgbm/src/test/scala/com/microsoft/azure/synapse/ml/lightgbm/split1/VerifyLightGBMCommon.scala b/lightgbm/src/test/scala/com/microsoft/azure/synapse/ml/lightgbm/split1/VerifyLightGBMCommon.scala index c4169cd5bd6..e6e0015dc31 100644 --- a/lightgbm/src/test/scala/com/microsoft/azure/synapse/ml/lightgbm/split1/VerifyLightGBMCommon.scala +++ b/lightgbm/src/test/scala/com/microsoft/azure/synapse/ml/lightgbm/split1/VerifyLightGBMCommon.scala @@ -8,7 +8,8 @@ import com.microsoft.azure.synapse.ml.lightgbm._ import com.microsoft.azure.synapse.ml.lightgbm.dataset.{ChunkedArrayUtils, SampledData} import com.microsoft.azure.synapse.ml.lightgbm.swig.{DoubleChunkedArray, DoubleSwigArray, IntSwigArray, SwigUtils} import com.microsoft.ml.lightgbm.{SWIGTYPE_p_p_void, SWIGTYPE_p_void, lightgbmlib} -import org.apache.spark.ml.linalg.{DenseVector, SparseVector} +import org.apache.spark.ml.attribute.{Attribute, AttributeGroup, NumericAttribute} +import org.apache.spark.ml.linalg.{DenseVector, SparseVector, Vectors} import org.apache.spark.sql.DataFrame // scalastyle:off magic.number @@ -18,6 +19,24 @@ class VerifyLightGBMCommon extends TestBase with LightGBMTestUtils { lazy val taskDF: DataFrame = loadBinary("task.train.csv", "TaskFailed10").cache() lazy val pimaDF: DataFrame = loadBinary("PimaIndian.csv", "Diabetes mellitus").cache() + /** Builds a tiny 4-row frame whose features vectors have `numFeatures` columns. */ + private def makeDuplicateNameDF(numFeatures: Int): DataFrame = { + val rows = Seq(0.0, 1.0, 0.0, 1.0).zipWithIndex.map { case (label, row) => + (label, Vectors.dense(Array.tabulate(numFeatures)(col => (row + col + 1).toDouble))) + } + spark.createDataFrame(rows).toDF(labelCol, featuresCol) + } + + /** A fresh minimal classifier per call, so slot names from one test cannot leak into another. */ + private def duplicateNameModel: LightGBMClassifier = new LightGBMClassifier() + .setFeaturesCol(featuresCol) + .setLabelCol(labelCol) + .setDefaultListenPort(getAndIncrementPort()) + .setNumLeaves(5) + .setNumIterations(5) + .setObjective("binary") + .setDataTransferMode(LightGBMConstants.StreamingDataTransferMode) + lazy val baseModel: LightGBMClassifier = new LightGBMClassifier() .setFeaturesCol(featuresCol) .setRawPredictionCol(rawPredCol) @@ -294,4 +313,73 @@ class VerifyLightGBMCommon extends TestBase with LightGBMTestUtils { (conv(up.last) + conv(down.head)) / fromInt(2) } } + + test("Verify duplicate feature names are handled correctly") { + // Regression test: LightGBM rejects a Dataset whose feature names repeat, failing with + // "Feature (Column_) appears more than one time". Spark can surface repeated names through + // AttributeGroup metadata on the features column, so SynapseML de-duplicates them first. + val attrs: Array[Attribute] = Array( + NumericAttribute.defaultAttr.withName("Column_").withIndex(0), + NumericAttribute.defaultAttr.withName("Column_").withIndex(1), + NumericAttribute.defaultAttr.withName("Column_").withIndex(2), + NumericAttribute.defaultAttr.withName("unique_col").withIndex(3)) + val attrGroup = new AttributeGroup(featuresCol, attrs) + + val df = makeDuplicateNameDF(4) + val dfWithDuplicateNames = df.withColumn( + featuresCol, + df(featuresCol).as(featuresCol, attrGroup.toMetadata())) + + val predictions = duplicateNameModel.fit(dfWithDuplicateNames).transform(dfWithDuplicateNames) + assert(predictions.count() == 4) + } + + test("Verify explicit slotNames parameter is used") { + val df = makeDuplicateNameDF(3) + val model = duplicateNameModel.setSlotNames(Array("feature_a", "feature_b", "feature_c")) + assert(model.fit(df).transform(df).count() == 4) + } + + test("Verify duplicate explicit slotNames are made unique") { + val df = makeDuplicateNameDF(3) + val model = duplicateNameModel.setSlotNames(Array("Column_", "Column_", "Column_")) + assert(model.fit(df).transform(df).count() == 4) + } + + test("Verify a generated slot name cannot collide with a later original name") { + // "Column_" repeats, so the second occurrence is renamed. A naive implementation renames it + // to "Column__1", which is already taken by the third slot, so LightGBM still fails with + // "Feature (Column__1) appears more than one time". The renamed slot must skip past every + // original name, not just the ones seen so far. + val df = makeDuplicateNameDF(3) + val model = duplicateNameModel.setSlotNames(Array("Column_", "Column_", "Column__1")) + assert(model.fit(df).transform(df).count() == 4) + } + + test("Verify names differing only by space vs underscore are made unique") { + // LightGBM replaces spaces with underscores before checking for duplicates, so "a b" and + // "a_b" are the same feature natively and fail with "Feature (a_b) appears more than one + // time" even though the two strings differ in Scala. + val df = makeDuplicateNameDF(3) + val model = duplicateNameModel.setSlotNames(Array("a b", "a_b", "c")) + assert(model.fit(df).transform(df).count() == 4) + } + + test("Verify slotNames of the wrong length are skipped rather than read out of bounds") { + // LGBM_DatasetSetFeatureNames reads numCols entries from the array, so a short slotNames + // array is an out-of-bounds native read. slotNames is user-supplied and never length-checked + // upstream, so LightGBMDataset.setFeatureNames guards every dataset-naming path. Training + // proceeds with LightGBM's own generated names instead of crashing the executor. + val df = makeDuplicateNameDF(4) + val model = duplicateNameModel.setSlotNames(Array("only_one_name")) + assert(model.fit(df).transform(df).count() == 4) + } + + test("Verify slotNames of the wrong length are skipped in bulk mode too") { + val df = makeDuplicateNameDF(4) + val model = duplicateNameModel + .setDataTransferMode(LightGBMConstants.BulkDataTransferMode) + .setSlotNames(Array("a", "b")) + assert(model.fit(df).transform(df).count() == 4) + } }