#374 Incremental Ingestion

[yruslan]

Closes #374
Closes #421

This PR adds 'incremental' as a schedule type, and mechanisms for managing offsets (experimental).

Pramen version 1.10 introduces the concept of incremental ingestion. It allows running a pipeline multiple times a day
without reprocessing data that was already processed. In order to enable it, use incremental schedule when defining your
ingestion operation:

schedule = "incremental"

In order for the incremental ingestion to work you need to define a monotonically increasing field, called an offset.
Usually, this incremental field can be a counter, or a record creation timestamp. You need to define the offset field in
your source. The source should support incremental ingestion in order to use this mode.

offset.column {
  name = "created_at"
  type = "datetime"
}

Offset types available at the moment:

Type	Description
integral	Any integral type (short, int, long)
datetime	A datetime or timestamp fields
string	Only string / varchar(n) types.

Only ingestion jobs support incremental schedule at the moment. Incremental transformations and sinks are planned to be
available soon.

[github-actions]

Unit Test Coverage

Overall Project	84.16% -2.66%	🍏
Files changed	77.8%	❌

Module	Coverage
pramen:core Jacoco Report	84.93% -2.93%	❌

Files

Module	File	Coverage
pramen:core Jacoco Report	IdentityTransformer.scala	100%	🍏
	OffsetInfoParser.scala	100%	🍏
	TableReaderJdbcConfig.scala	100% -1.08%	🍏
	JournalHadoopCsv.scala	100%	🍏
	JdbcSource.scala	100%	🍏
	TableReaderJdbcBase.scala	100%	🍏
	JournalTasks.scala	100%	🍏
	TaskCompletedCsv.scala	100% -18.46%	❌
	JournalTask.scala	100% -6.15%	❌
	MetaTableStats.scala	100% -57.14%	❌
	OffsetRecords.scala	100%	🍏
	OffsetRecordConverter.scala	100%	🍏
	DataOffsetRequest.scala	100%	❌
	DataOffsetAggregated.scala	100%	❌
	OffsetRecord.scala	100%	❌
	MetaTable.scala	98.91%	🍏
	Schedule.scala	98.58% -8.87%	❌
	MetastorePersistenceRaw.scala	98.39%	🍏
	TransferTable.scala	97.83%	🍏
	TaskCompleted.scala	97.48% -7.56%	🍏
	InfoDateConfig.scala	97%	🍏
	TransferJob.scala	96.77%	🍏
	SqlGeneratorDenodo.scala	96.43% -1.34%	🍏
	SqlGeneratorDb2.scala	96.4% -1.35%	🍏
	SqlGeneratorHive.scala	96.38% -1.36%	🍏
	SqlGeneratorHsqlDb.scala	96.38% -1.36%	🍏
	OperationSplitter.scala	96.29%	🍏
	SqlGeneratorGeneric.scala	96.26% -1.4%	🍏
	ScheduleStrategyUtils.scala	95.78%	🍏
	TransformationJob.scala	95.56% -1.78%	🍏
	SqlGeneratorSas.scala	95.5% -1.74%	🍏
	SparkSource.scala	95.45% -4.03%	🍏
	LocalSparkSource.scala	94.88% -1.02%	❌
	SparkUtils.scala	94.76% -0.57%	🍏
	SinkJob.scala	93.67%	🍏
	SqlGeneratorOracle.scala	93.38% -1.32%	🍏
	PramenImpl.scala	93.23%	🍏
	ScheduleStrategyIncremental.scala	93.16% -6.84%	🍏
	SqlGeneratorPostgreSQL.scala	91.71% -5.99%	🍏
	SqlGeneratorMySQL.scala	91.71% -3.23%	🍏
	PythonTransformationJob.scala	91.71%	🍏
	PipelineNotificationBuilderHtml.scala	91.3% -1.74%	❌
	JobBase.scala	91.26% -1.82%	🍏
	AppContextImpl.scala	90.56% -2.58%	🍏
	MetastoreImpl.scala	90.33% -1.73%	🍏
	ConcurrentJobRunnerImpl.scala	88.98% -8.33%	❌
	AppRunner.scala	88.91%	🍏
	PipelineStateImpl.scala	88.89%	🍏
	OperationDef.scala	88.74% -1.2%	❌
	RawFileSource.scala	88.72% -0.51%	❌
	TaskRunnerBase.scala	88.47% -0.17%	🍏
	SqlGeneratorMicrosoft.scala	87.5% -2.53%	🍏
	OffsetManagerJdbc.scala	86.85% -13.15%	🍏
	MetastorePersistenceParquet.scala	86.6% -2.49%	🍏
	MetastorePersistence.scala	84.43%	🍏
	MetastorePersistenceTransient.scala	84.38%	🍏
	TableReaderJdbc.scala	84.03% -2.47%	❌
	IncrementalIngestionJob.scala	81.65% -18.35%	🍏
	JournalJdbc.scala	81.53%	🍏
	BookkeeperBase.scala	81.21% -3.76%	❌
	IngestionJob.scala	80.93%	🍏
	TableReaderJdbcNative.scala	78.96% -16.69%	❌
	MetastorePersistenceDelta.scala	71.7% -11.52%	❌
	TableReaderSpark.scala	70.63% -25.92%	❌
	Bookkeeper.scala	69.28%	🍏
	SlickUtils.scala	67.05% -13.64%	❌
	MetastorePersistenceTransientEager.scala	64.39% -7.2%	❌
	BookkeeperJdbc.scala	56.35% -0.73%	❌
	PramenDb.scala	54.19% -1.83%	❌

[yruslan]

Putting this back to draft since Kevin suggested to use full intervals for offset tracking instead of half-intervals.
Potential benefits:

• Easier to understand the offsets table
• We don't have to rely on absolute minimum values for each offset type

[kevinwallimann]

Should there not be the same logic as in line 75? I.e. empty seq if maximumInfoDate is after infoDate?

[yruslan]

This is because when we don't have an information date at the source, our information date is the processing date, and we can't run for previous information dates, hence empty Seq. But when the source has an information column, new offsets could have arrived for that specific date, so we allow it to run.

When I looked at the code, and especially when started writing unit tests for it as you suggested, I decided to refactor it a bit to make it less confising.

[kevinwallimann]

There seems to be some duplicated logic with getOffsets. What different purpose do these two methods have?

[yruslan]

The number offsets for a table could be quite a lot, but uncommitted offsets count should be very low. So this method makes sure that when we get uncommitted offsets for a table, the filtering is done on the database, not after they are transferred.

getOffsets() requires an information date to ensure millions of offsets won't be loaded, but returns both committed and uncommitted offsets.

[kevinwallimann]

Why did you not test the splitComplexIdentifier for mysql, but for MicrosoftSQL?

[yruslan]

splitComplexIdentifier() is defined in SqlGeneratorBase which is tested already in SqlGeneratorDenodoSuite. But for MicrosoftSQL the method is special in order to support 2 types of escaping [] and "" and its mix.

But I'll move testing of splitComplexIdentifier() defined at SqlGeneratorBase to SqlGeneratorGenericSuite or create SqlGeneratorBaseSuite to avoid confusion

[kevinwallimann]

Why did you not add a test for quote and unquote?

[yruslan]

Added tests. And found a bug. 😄

[yruslan]

Hi Kevin, I've fixed all obvious issues, and added explanations for some. I'm happy with the results. Thanks again for the effort reviewing this!

Going to merge this now to unblock some other issues on the stack. Comments remain open, and if you want to continue conversation on some of them will be happy to.