import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
    * Calculate the number of step copies in a step.<br>
    * If a step is not running clustered, it's simply returning getCopies().<br>
    * If a step is clustered and not doing any partitioning, it's simply returning getCopies().<br>
    * If a step is clustered and partitioned, we need to look in the partitioning map for the specified slave server.<br>
    * That is because the number of copies can vary over the slaves. (5 partitions over 3 slaves for example)
    * 
    * @param slaveServer the slave server
    * @param referenceStep the reference step
    * @return the number of step copies that we run.
    */
   private int determineNrOfStepCopies(SlaveServer slaveServer, StepMeta step) {
	if (!step.isClustered()) return step.getCopies();
	if (!step.isPartitioned()) return step.getCopies();
	if (slaveServer.isMaster()) return step.getCopies();
	
	// Partitioned and clustered...
	//
	StepPartitioningMeta stepPartitioningMeta = step.getStepPartitioningMeta();
	PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
	
	Map<PartitionSchema, List<String>> partitionMap = slaveServerPartitionsMap.get(slaveServer);
	List<String> partitionList = partitionMap.get(partitionSchema);
	
	return partitionList.size();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
    * Create a copy of a step from the original transformation for use in the a slave transformation.
    * If the step is partitioned, the partitioning will be changed to "schemaName (slave)" 
    * 
    * @param stepMeta The step to copy / clone.
    * @return a copy of the specified step for use in a slave transformation.
    */
   private StepMeta addSlaveCopy(TransMeta transMeta, StepMeta stepMeta, SlaveServer slaveServer) {
	StepMeta copy = (StepMeta) stepMeta.clone();
	if (copy.isPartitioned()) {
		StepPartitioningMeta stepPartitioningMeta = copy.getStepPartitioningMeta();
		PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
		String slavePartitionSchemaName = createSlavePartitionSchemaName(partitionSchema.getName());
		PartitionSchema slaveSchema = transMeta.findPartitionSchema(slavePartitionSchemaName);
		if (slaveSchema!=null) {
			stepPartitioningMeta.setPartitionSchema(slaveSchema);
		}
		// Always just start a single copy on the slave server...
		// Otherwise the confusion w.r.t. to partitioning & re-partitioning would be complete.
		//
		copy.setCopies(1);
	}

	// Remove the clustering information on the slave transformation step
	// We don't need it anymore, it only confuses.
	//
	copy.setClusterSchema(null); 
	
	transMeta.addStep(copy);
	return copy;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public boolean isUsingPartitionSchema(PartitionSchema partitionSchema)
{
    // Loop over all steps and see if the partition schema is used.
    for (int i=0;i<nrSteps();i++)
    {
        StepPartitioningMeta stepPartitioningMeta = getStep(i).getStepPartitioningMeta();
        if (stepPartitioningMeta!=null)
        {
            PartitionSchema check = stepPartitioningMeta.getPartitionSchema();
            if (check!=null && check.equals(partitionSchema))
            {
                return true;
            }
        }
    }
    return false;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
   * Calculate the number of step copies in a step.<br>
   * If a step is not running clustered, it's simply returning getCopies().<br>
   * If a step is clustered and not doing any partitioning, it's simply returning getCopies().<br>
   * If a step is clustered and partitioned, we need to look in the partitioning map for the specified slave server.<br>
   * That is because the number of copies can vary over the slaves. (5 partitions over 3 slaves for example)
   * 
   * @param slaveServer the slave server
   * @param referenceStep the reference step
   * @return the number of step copies that we run.
   */
  private int determineNrOfStepCopies(SlaveServer slaveServer, StepMeta step) {
  if (!step.isClustered()) return step.getCopies();
  if (!step.isPartitioned()) return step.getCopies();
  if (slaveServer.isMaster()) return step.getCopies();
  
  // Partitioned and clustered...
  //
  StepPartitioningMeta stepPartitioningMeta = step.getStepPartitioningMeta();
  PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
  
  Map<PartitionSchema, List<String>> partitionMap = slaveServerPartitionsMap.get(slaveServer);
  List<String> partitionList = partitionMap.get(partitionSchema);
  
  return partitionList.size();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
   * Create a copy of a step from the original transformation for use in the a slave transformation.
   * If the step is partitioned, the partitioning will be changed to "schemaName (slave)" 
   * 
   * @param stepMeta The step to copy / clone.
   * @return a copy of the specified step for use in a slave transformation.
   */
  private StepMeta addSlaveCopy(TransMeta transMeta, StepMeta stepMeta, SlaveServer slaveServer) {
  StepMeta copy = (StepMeta) stepMeta.clone();
  if (copy.isPartitioned()) {
    StepPartitioningMeta stepPartitioningMeta = copy.getStepPartitioningMeta();
    PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
    String slavePartitionSchemaName = createSlavePartitionSchemaName(partitionSchema.getName());
    PartitionSchema slaveSchema = transMeta.findPartitionSchema(slavePartitionSchemaName);
    if (slaveSchema!=null) {
      stepPartitioningMeta.setPartitionSchema(slaveSchema);
    }
    // Always just start a single copy on the slave server...
    // Otherwise the confusion w.r.t. to partitioning & re-partitioning would be complete.
    //
    copy.setCopies(1);
  }

  // Remove the clustering information on the slave transformation step
  // We don't need it anymore, it only confuses.
  //
  copy.setClusterSchema(null); 
  
  transMeta.addStep(copy);
  return copy;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * Checks if the transformation is using the specified partition schema.
 *
 * @param partitionSchema the partition schema
 * @return true if the transformation is using the partition schema, false otherwise
 */
public boolean isUsingPartitionSchema(PartitionSchema partitionSchema)
{
    // Loop over all steps and see if the partition schema is used.
    for (int i=0;i<nrSteps();i++)
    {
        StepPartitioningMeta stepPartitioningMeta = getStep(i).getStepPartitioningMeta();
        if (stepPartitioningMeta!=null)
        {
            PartitionSchema check = stepPartitioningMeta.getPartitionSchema();
            if (check!=null && check.equals(partitionSchema))
            {
                return true;
            }
        }
    }
    return false;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * Calculate the number of step copies in a step.<br>
 * If a step is not running clustered, it's simply returning getCopies().<br>
 * If a step is clustered and not doing any partitioning, it's simply returning getCopies().<br>
 * If a step is clustered and partitioned, we need to look in the partitioning map for the specified slave server.<br>
 * That is because the number of copies can vary over the slaves. (5 partitions over 3 slaves for example)
 *
 * @param slaveServer
 *          the slave server
 * @param step
 *          the reference step
 * @return the number of step copies that we run.
 */
private int determineNrOfStepCopies( SlaveServer slaveServer, StepMeta step ) {
  if ( !step.isClustered() ) {
    return step.getCopies();
  }
  if ( !step.isPartitioned() ) {
    return step.getCopies();
  }
  if ( slaveServer.isMaster() ) {
    return step.getCopies();
  }

  // Partitioned and clustered...
  //
  StepPartitioningMeta stepPartitioningMeta = step.getStepPartitioningMeta();
  PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();

  Map<PartitionSchema, List<String>> partitionMap = slaveServerPartitionsMap.get( slaveServer );
  List<String> partitionList = partitionMap.get( partitionSchema );

  return partitionList.size();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * Create a copy of a step from the original transformation for use in the a slave transformation. If the step is
 * partitioned, the partitioning will be changed to "schemaName (slave)"
 *
 * @param stepMeta
 *          The step to copy / clone.
 * @return a copy of the specified step for use in a slave transformation.
 */
private StepMeta addSlaveCopy( TransMeta transMeta, StepMeta stepMeta, SlaveServer slaveServer ) {
  StepMeta copy = (StepMeta) stepMeta.clone();
  if ( copy.isPartitioned() ) {
    StepPartitioningMeta stepPartitioningMeta = copy.getStepPartitioningMeta();
    PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
    String slavePartitionSchemaName = createSlavePartitionSchemaName( partitionSchema.getName() );
    PartitionSchema slaveSchema = transMeta.findPartitionSchema( slavePartitionSchemaName );
    if ( slaveSchema != null ) {
      stepPartitioningMeta.setPartitionSchema( slaveSchema );
    }
    // Always just start a single copy on the slave server...
    // Otherwise the confusion w.r.t. to partitioning & re-partitioning would be complete.
    //
    copy.setCopies( 1 );
  }

  // Remove the clustering information on the slave transformation step
  // We don't need it anymore, it only confuses.
  //
  copy.setClusterSchema( null );

  transMeta.addStep( copy );
  return copy;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * This case simulates when we do have 2 step partitioned with one same partitioner We want to get a 'swim-lanes'
 * transformation
 * 
 * @throws KettlePluginException
 */
private void prepareStepMetas_cl1_cl1() throws KettlePluginException {
  StepMeta dummy1 = new StepMeta( ONE, null );
  StepMeta dummy2 = new StepMeta( TWO, null );

  PartitionSchema schema = new PartitionSchema( "p1", Arrays.asList( new String[] { PID1, PID2 } ) );
  // for delayed binding StepPartitioning meta does not achieve
  // schema name when using in constructor so we have to set it
  // explicitly. See equals implementation for StepPartitioningMeta.
  StepPartitioningMeta partMeta = new StepPartitioningMeta( "Mirror to all partitions", schema );
  // that is what I am talking about:
  partMeta.setPartitionSchemaName( schema.getName() );

  dummy1.setStepPartitioningMeta( partMeta );
  dummy2.setStepPartitioningMeta( partMeta );

  chain.add( dummy1 );
  chain.add( dummy2 );
  for ( StepMeta item : chain ) {
    item.setStepMetaInterface( new DummyTransMeta() );
  }
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * This is a case when we have 2 steps, but partitioned differently
 * 
 * @throws KettlePluginException
 */
private void prepareStepMetas_cl1_cl2() throws KettlePluginException {
  StepMeta dummy1 = new StepMeta( ONE, null );
  StepMeta dummy2 = new StepMeta( TWO, null );

  PartitionSchema schema1 = new PartitionSchema( "p1", Arrays.asList( new String[] { PID1, PID2 } ) );
  PartitionSchema schema2 = new PartitionSchema( "p2", Arrays.asList( new String[] { PID1, PID2 } ) );

  StepPartitioningMeta partMeta1 = new StepPartitioningMeta( "Mirror to all partitions", schema1 );
  StepPartitioningMeta partMeta2 = new StepPartitioningMeta( "Mirror to all partitions", schema2 );
  partMeta1.setPartitionSchemaName( schema1.getName() );
  partMeta2.setPartitionSchemaName( schema2.getName() );

  dummy1.setStepPartitioningMeta( partMeta1 );
  dummy2.setStepPartitioningMeta( partMeta2 );

  chain.add( dummy1 );
  chain.add( dummy2 );
  for ( StepMeta item : chain ) {
    item.setStepMetaInterface( new DummyTransMeta() );
  }
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * This is a case when first step running 2 copies and next is partitioned one.
 * 
 * @throws KettlePluginException
 */
private void prepareStepMetas_x2_cl1() throws KettlePluginException {
  StepMeta dummy1 = new StepMeta( ONE, null );
  StepMeta dummy2 = new StepMeta( TWO, null );

  PartitionSchema schema1 = new PartitionSchema( "p1", Arrays.asList( new String[] { PID1, PID2 } ) );
  StepPartitioningMeta partMeta1 = new StepPartitioningMeta( "Mirror to all partitions", schema1 );

  dummy2.setStepPartitioningMeta( partMeta1 );
  dummy1.setCopies( 2 );

  chain.add( dummy1 );
  chain.add( dummy2 );
  for ( StepMeta item : chain ) {
    item.setStepMetaInterface( new DummyTransMeta() );
  }
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public ModPartitionerDialog(Shell parent, Object in, StepPartitioningMeta partitioningMeta, TransMeta transMeta)
{
	super(parent, (BaseStepMeta)in, transMeta, partitioningMeta.getPartitioner().getDescription() );
	this.partitioningMeta = partitioningMeta;
	partitioner = (ModPartitioner) partitioningMeta.getPartitioner();
	stepMeta=(StepMetaInterface)in;
	fieldName = partitioner.getFieldName();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public StepDialogInterface getPartitionerDialog(Object stepMeta, 
			StepPartitioningMeta partitioningMeta, TransMeta transMeta) throws KettleException
{

	String dialogClassName = partitioningMeta.getPartitioner().getDialogClassName();

	Class<?> dialogClass;
	Class<?>[] paramClasses = new Class[] { Shell.class, Object.class, StepPartitioningMeta.class, TransMeta.class };
	Object[] paramArgs = new Object[] { spoon.getShell(), stepMeta, partitioningMeta, transMeta };
	Constructor<?> dialogConstructor;
	try
	{
		dialogClass = partitioningMeta.getClass().getClassLoader().loadClass(dialogClassName);
		dialogConstructor = dialogClass.getConstructor(paramClasses);
		return (StepDialogInterface) dialogConstructor.newInstance(paramArgs);
	} catch (Exception e)
	{
		// try the old way for compatibility 
		Method method = null;
			try {
				Class<?> sig[] = new Class[] {Shell.class, StepMetaInterface.class, TransMeta.class};
				method = stepMeta.getClass().getDeclaredMethod( "getDialog", sig );
				if( method != null ) {
					return (StepDialogInterface) method.invoke( stepMeta, new Object[] { spoon.getShell(), stepMeta, transMeta } );
				}
			} catch (Throwable t) {
			}

		throw new KettleException(e);
	}

}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public StepPartitioningMeta loadStepPartitioningMeta(ObjectId id_step) throws KettleException
{
	StepPartitioningMeta stepPartitioningMeta = new StepPartitioningMeta();
	
	stepPartitioningMeta.setPartitionSchemaName( repository.getStepAttributeString(id_step, "PARTITIONING_SCHEMA") );
    String methodCode   = repository.getStepAttributeString(id_step, "PARTITIONING_METHOD");
    stepPartitioningMeta.setMethod( StepPartitioningMeta.getMethod(methodCode) );
    if( stepPartitioningMeta.getPartitioner() != null ) {
    	stepPartitioningMeta.getPartitioner().loadRep( repository, id_step);
    }
    stepPartitioningMeta.hasChanged(true);
    
    return stepPartitioningMeta;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public ModPartitionerDialog(Shell parent, StepMeta stepMeta, StepPartitioningMeta partitioningMeta, TransMeta transMeta)
{
	super(parent, (BaseStepMeta)stepMeta.getStepMetaInterface(), transMeta, partitioningMeta.getPartitioner().getDescription() );
	this.stepMeta = stepMeta;
	this.partitioningMeta = partitioningMeta;
	partitioner = (ModPartitioner) partitioningMeta.getPartitioner();
	fieldName = partitioner.getFieldName();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public StepDialogInterface getPartitionerDialog(StepMeta stepMeta, StepPartitioningMeta partitioningMeta, TransMeta transMeta) throws KettleException
{
	Partitioner partitioner = partitioningMeta.getPartitioner();
	String dialogClassName = partitioner.getDialogClassName();

	Class<?> dialogClass;
	Class<?>[] paramClasses = new Class[] { Shell.class, StepMeta.class, StepPartitioningMeta.class, TransMeta.class };
	Object[] paramArgs = new Object[] { spoon.getShell(), stepMeta, partitioningMeta, transMeta };
	Constructor<?> dialogConstructor;
	try
	{
		dialogClass = partitioner.getClass().getClassLoader().loadClass(dialogClassName);
		dialogConstructor = dialogClass.getConstructor(paramClasses);
		return (StepDialogInterface) dialogConstructor.newInstance(paramArgs);
	} catch (Exception e)
	{
		// try the old way for compatibility 
		Method method = null;
			try {
				Class<?> sig[] = new Class[] {Shell.class, StepMetaInterface.class, TransMeta.class};
				method = stepMeta.getClass().getDeclaredMethod( "getDialog", sig );
				if( method != null ) {
					return (StepDialogInterface) method.invoke( stepMeta, new Object[] { spoon.getShell(), stepMeta, transMeta } );
				}
			} catch (Throwable t) {
			}

		throw new KettleException(e);
	}

}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
public StepPartitioningMeta loadStepPartitioningMeta( ObjectId id_step ) throws KettleException {
  StepPartitioningMeta stepPartitioningMeta = new StepPartitioningMeta();

  stepPartitioningMeta.setPartitionSchemaName( repository
    .getStepAttributeString( id_step, "PARTITIONING_SCHEMA" ) );
  String methodCode = repository.getStepAttributeString( id_step, "PARTITIONING_METHOD" );
  stepPartitioningMeta.setMethod( StepPartitioningMeta.getMethod( methodCode ) );
  if ( stepPartitioningMeta.getPartitioner() != null ) {
    stepPartitioningMeta.getPartitioner().loadRep( repository, id_step );
  }
  stepPartitioningMeta.hasChanged( true );

  return stepPartitioningMeta;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * Checks if the transformation is using the specified partition schema.
 *
 * @param partitionSchema
 *          the partition schema
 * @return true if the transformation is using the partition schema, false otherwise
 */
public boolean isUsingPartitionSchema( PartitionSchema partitionSchema ) {
  // Loop over all steps and see if the partition schema is used.
  for ( int i = 0; i < nrSteps(); i++ ) {
    StepPartitioningMeta stepPartitioningMeta = getStep( i ).getStepPartitioningMeta();
    if ( stepPartitioningMeta != null ) {
      PartitionSchema check = stepPartitioningMeta.getPartitionSchema();
      if ( check != null && check.equals( partitionSchema ) ) {
        return true;
      }
    }
  }
  return false;
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
/**
 * Row listeners collection modifiers are exposed out of BaseStep class,
 * whereas the collection traversal is happening on every row being processed.
 *
 * We should be sure that modification of the collection will not throw a concurrent modification exception.
 */
@Test
public void testRowListeners() throws Exception {
    int modifiersAmount = 100;
    int traversersAmount = 100;

    StepMeta stepMeta = mock( StepMeta.class);
    TransMeta transMeta = mock( TransMeta.class);
    when( stepMeta.getName() ).thenReturn( STEP_META );
    when( transMeta.findStep( STEP_META ) ).thenReturn( stepMeta );
    when( stepMeta.getTargetStepPartitioningMeta() ).thenReturn( mock( StepPartitioningMeta.class ) );

    baseStep = new BaseStep( stepMeta, null, 0, transMeta, mock( Trans.class ) );

    AtomicBoolean condition = new AtomicBoolean( true );

    List<Modifier> modifiers = new ArrayList<>();
    for ( int i = 0; i < modifiersAmount; i++ ) {
        modifiers.add( new Modifier( condition ) );
    }
    List<Traverser> traversers = new ArrayList<>();
    for ( int i = 0; i < traversersAmount; i++ ) {
        traversers.add( new Traverser( condition ) );
    }

    ConcurrencyTestRunner<?, ?> runner =
            new ConcurrencyTestRunner<Object, Object>( modifiers, traversers, condition );
    runner.runConcurrentTest();

    runner.checkNoExceptionRaised();
}
 

import org.pentaho.di.trans.step.StepPartitioningMeta; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  databaseMeta = mock( DatabaseMeta.class );
  doReturn( "" ).when( databaseMeta ).quoteField( anyString() );

  dimensionLookupMeta = mock( DimensionLookupMeta.class );
  doReturn( databaseMeta ).when( dimensionLookupMeta ).getDatabaseMeta();
  doReturn( new String[]{} ).when( dimensionLookupMeta ).getKeyLookup();
  doReturn( new String[]{} ).when( dimensionLookupMeta ).getFieldLookup();
  doReturn( new int[]{} ).when( dimensionLookupMeta ).getFieldUpdate();

  stepMeta = mock( StepMeta.class );
  doReturn( "step" ).when( stepMeta ).getName();
  doReturn( mock( StepPartitioningMeta.class ) ).when( stepMeta ).getTargetStepPartitioningMeta();
  doReturn( dimensionLookupMeta ).when( stepMeta ).getStepMetaInterface();

  Database db = mock( Database.class );
  doReturn( mock( Connection.class ) ).when( db ).getConnection();

  dimensionLookupData = mock( DimensionLookupData.class );
  dimensionLookupData.db = db;
  dimensionLookupData.keynrs = new int[] { };
  dimensionLookupData.fieldnrs = new int[] { };

  TransMeta transMeta = mock( TransMeta.class );
  doReturn( stepMeta ).when( transMeta ).findStep( anyString() );

  dimensionLookup = new DimensionLookup( stepMeta, dimensionLookupData, 1, transMeta, mock( Trans.class ) );
  dimensionLookup.setData( dimensionLookupData );
  dimensionLookup.setMeta( dimensionLookupMeta );
  dimensionLookupSpy = spy( dimensionLookup );
  doReturn( stepMeta ).when( dimensionLookupSpy ).getStepMeta();
  doReturn( false ).when( dimensionLookupSpy ).isRowLevel();
  doReturn( false ).when( dimensionLookupSpy ).isDebug();
  doReturn( true ).when( dimensionLookupSpy ).isAutoIncrement();
  doNothing().when( dimensionLookupSpy ).logDetailed( anyString() );
}