Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Feat/5.0 changes #1008

Merged
merged 7 commits into from
Nov 30, 2023
Merged

Feat/5.0 changes #1008

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f74590a
An issue was present in the S3 sink handling null value. This was lea…
Nov 29, 2023
a553fd8
The PR introduces the support for S3 sink for BigDecimal. A new case…
Nov 29, 2023
f8a920b
Initialised the bigdecimal using string literal
Nov 30, 2023
13680cd
Fix the integration tests compilation errors
Nov 30, 2023
d28ab70
Fix the integration tests compilation errors
Nov 30, 2023
49f3c63
Set the S3 uri used in the test, or otherwise the test fails
Nov 30, 2023
ad82b9c
Merge branch 'master' into feat/5.0-changes
stheppi Nov 30, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 63 additions & 0 deletions ...s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3AvroWriterManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,69 @@ class S3AvroWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont

}

"avro sink" should "write BigDecimal" in {
val sink = S3WriterManager.from(avroConfig)
val usersWithDecimal1 =
new Struct(UsersSchemaDecimal)
.put("name", "sam")
.put("title", "mr")
.put(
"salary",
BigDecimal(100.43).setScale(18).bigDecimal,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's advisable to get into the habit of constructing using BigDecimal(String) instead of relying on floating-point literals. The conversion of floating-point literals, such as 100.43, to BigDecimal using setScale(18) may introduce rounding errors due to the binary representation of these literals. To ensure accurate representation and avoid precision loss, consider using BigDecimal("100.43").setScale(18) instead.

Additionally, it's worth evaluating whether a scale of 18 is necessary for the specific use case, as an excessively high scale might impact performance and memory usage without providing significant benefits.

(This comment applies to various places in the PR.)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for raising that. The 18 scale comes from the schema definition. They should align or Avro serialisation will fail.

)
val writeRes1 = sink.write(
TopicPartitionOffset(Topic(TopicName), 1, Offset(1L)),
MessageDetail(
NullSinkData(None),
StructSinkData(usersWithDecimal1),
Map.empty,
None,
Topic(TopicName),
1,
Offset(1L),
),
)

writeRes1.isRight should be(true)

val usersWithDecimal2 =
new Struct(UsersSchemaDecimal)
.put("name", "maria")
.put("title", "ms")
.put(
"salary",
BigDecimal(100.43).setScale(18).bigDecimal,
)

val writeRes2 = sink.write(
TopicPartitionOffset(Topic(TopicName), 1, Offset(2L)),
MessageDetail(
NullSinkData(None),
StructSinkData(usersWithDecimal2),
Map.empty,
None,
Topic(TopicName),
1,
Offset(2L),
),
)
writeRes2.isRight should be(true)
sink.close()

listBucketPath(BucketName, "streamReactorBackups/myTopic/1/").size should be(1)

val byteArray = remoteFileAsBytes(BucketName, "streamReactorBackups/myTopic/1/2.avro")
val genericRecords: List[GenericRecord] = avroFormatReader.read(byteArray)
genericRecords.size should be(2)

genericRecords(0).get("name").toString should be("sam")
val bb1 = genericRecords(0).get("salary").asInstanceOf[ByteBuffer]
Decimal.toLogical(Decimal.builder(18).optional().build(), bb1.array()) should be(
BigDecimal(100.43).setScale(18).bigDecimal,
)

}

"avro sink" should "write start a new file in case of schema change" in {

val secondSchema: Schema = SchemaBuilder.struct()
Expand Down
70 changes: 70 additions & 0 deletions ...s3/src/it/scala/io/lenses/streamreactor/connect/aws/s3/sink/S3JsonWriterManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -176,4 +176,74 @@ class S3JsonWriterManagerTest extends AnyFlatSpec with Matchers with S3ProxyCont
)
}

"json sink" should "write bigdecimal to json" in {

val bucketAndPrefix = S3Location(BucketName, PathPrefix.some)
val config = S3SinkConfig(
S3Config(
None,
Some(Identity),
Some(Credential),
AuthMode.Credentials,
),
bucketOptions = Seq(
SinkBucketOptions(
TopicName.some,
bucketAndPrefix,
commitPolicy = CommitPolicy(Count(1)),
formatSelection = JsonFormatSelection,
keyNamer = new S3KeyNamer(
AvroFormatSelection,
defaultPartitionSelection(Values),
new OffsetS3FileNamer(
identity[String],
JsonFormatSelection.extension,
),
new PaddingService(Map[String, PaddingStrategy](
"partition" -> NoOpPaddingStrategy,
"offset" -> LeftPadPaddingStrategy(12, 0),
)),
),
localStagingArea = LocalStagingArea(localRoot),
partitionSelection = defaultPartitionSelection(Values),
dataStorage = DataStorageSettings.disabled,
),
),
offsetSeekerOptions = OffsetSeekerOptions(5),
compressionCodec,
batchDelete = true,
)

val sink = S3WriterManager.from(config)

val topic = Topic(TopicName)
val offset = Offset(1L)
val usersWithDecimal =
new Struct(UsersSchemaDecimal)
.put("name", "sam")
.put("title", "mr")
.put(
"salary",
BigDecimal(100.43).setScale(18).bigDecimal,
)
sink.write(
TopicPartitionOffset(topic, 1, offset),
MessageDetail(NullSinkData(None),
StructSinkData(usersWithDecimal),
Map.empty[String, SinkData],
None,
topic,
0,
offset,
),
)

sink.close()

listBucketPath(BucketName, "streamReactorBackups/myTopic/1/").size should be(1)

remoteFileAsString(BucketName, "streamReactorBackups/myTopic/1/1.json") should be(
s"""{"name":"sam","title":"mr","salary":100.430000000000000000}""",
)
}
}
Loading