Handling Kinesis Stream Event Batching with Typesense

Hi everyone!

So I have a Kinesis stream with a really big amount of indexes that should be created/updated/deleted.

How would I go about batching events so that I can be agnostic about the action? From the documentation, I've seen that one can batch an array of objects with one action:

client.collections('companies').documents().import(documents, {action: 'create'})

My main question is, would it be possible to make a batch with different actions or will I be forced to make different batches depending on the actions of the data that I receive from the kinesis stream?

For example, Algollia has this:

client.multipleBatch([
  { action: 'addObject', indexName: 'index1', body: { firstname: 'Jimmie', lastname: 'Barninger' } },
  { action: 'deleteObject', indexName: 'index2', body: { objectID: 'myID5' } }
])

👋 Dui We don't have a universal actions end-point. The import end-point has an upsert mode that you can use for creation/update.

Deletes have to be sent separately. But we don't have a batched delete end-point for that.

We're aware of this limitation and hope to address it in the future.

Thanks for such a quick answer!

Okay, so I'm gonna have to make a few different batches depending on the action then?
You wrote: "We're aware of this limitation and hope to address it in the future." Do you mean a universal actions end-point or batched delete -endpoint?

We want to explore an universal action end-point, but in the mean time, instead of different batches, logical deletion is the easiest approach.

You can have a single Kinesis consumer that can handle insert, update and delete.

Do you see any problem with using logical deletes?

Nice!

So you would recommend that my kinesis consumer would differentiate between the action and run an import(action) for each?

Logical deletion seems fine as for now, I guess the only downside is that I'd have to massage the data a little bit to be able to know which ones to delete, but it's basically how we are working today anyway 🙂

You can just use action=upsert regardless of whether the record is an insert or update or delete and send them all in the same batch, in a single API call.

The only additional processing you need to do is for the deletion case:. You can send a simple {"id": "<id>", "is_deleted: true} document or modify the actual document (if that's available to the consumer) to set the is_deleted field.

Ah, of course! Thanks 🙂

Do they run parallel? And what version of Typesense are you using?

I'm using v1.0.2 and yes, they are two separate import calls. I didn't think I could batch two different actions in one import call?

You mean 0.21? The Typesense server version, not the client.

ah, right: v0.21.0

In v0.21, the import jobs were multiplexed because we didn't think that people might run parallel updates to the same collection. In v0.22 we have fixed that so that parallel writes to a given collection is always serialized.

Ok, so my problem will persist until I update to the upcoming v0.22?

> I didn't think I could batch two different actions in one import call?
That's correct, unless you always had the whole document, in which case, it's just upsert.

Oh I'm using the cloud version

No problem, we can upgrade your Cloud version to the latest stable 0.22 RC version.

Awesome. Are there docs for it?

Give us a few hours, and we will update this thread once that's done.

sweet! Thc

Here are the docs: https://github.com/typesense/typesense-website/blob/v0.22.0-docs/docs-site/content/0.22.0/api/README.md

> No problem, we can upgrade your Cloud version to the latest stable 0.22 RC version.
Do you need anything from me?

Yes, your cluster ID. You can email us [email protected]

Hey btw! Will nested objects be available from 0.22 as well?

No, we started work on that but had to defer since the changes were more involved.

The cluster has been upgraded.

Thanks a bunch! One weird thing that remains.

I make two updates in one import:

const body = [
  { id: 'test_id', lastChance: false },
  {
     id: 'test_id',
     'price_SE.amount': 5000,
     'price_SE.currency': 'SEK'
  }
]

typesenseClient
  .collections('test_index')
  .documents()
  .import(body, { 'update', batch_size: 100 })

When I look in the data in typesense, it seems like the lastChance-field from the first object is undefined. And if I reverse the order of these two, the price_SE.amount gets undefined instead.

Do you know what could be the case? It's like it is updating the fields that are missing from each of the objects with undefined instead of ignoring them.

What was the original field value of the test_id document before the update was called?

Ok, thanks! Probably missing something with the initial value and that it is optional

> When I look in the data in typesense, it seems like the lastChance-field from the first object is undefined.
Does the field not exist in the returned document, or the value of the field is undefined? I won't rule out a bug lurking here, but need more information to ascertain that.

it doesn't exist in the first object, that's why I thought it would be ignored

You first inserted a document with no lastChance field, then ran the import above and now when you query the document the lastChance is not found? Sorry for being a bit slow here 🙂

yeah so I ran an upsert import with no lastChance . After that, I ran two update imports where one of them contain lastChance. Then the lastChance is not found.

That shouldn't happen. Can you please share a code snippet (any client is fine) which reproduces the issue in a small stand-alone example that I can run?

this is my code:

const batchImport = async ({ index, action, body }) => {
  if (!body) return
  typesenseClient
    .collections(index)
    .documents()
    .import(body, { action, batch_size: 100 })
    .then((res) => console.log(JSON.stringify(res)))
    .catch((e) => console.error(JSON.stringify(e)))
}

await batchImport({ index, action: 'upsert', body: groupedData['upsert'] })
await batchImport({ index, action: 'update', body: groupedData['update'] })

this is the data:

groupedData {
  upsert: [
    {
      id: 'CYgQsv4oZi',
      'metadata.material': [Array],
      'metadata.brand': 'Nike',
      'categories.lvl0': [Array],
      'categories.lvl1': [Array],
      'categories.lvl2': [Array],
      'categories.lvl3': [Array]
    }
  ],
  update: [
    { id: 'CYgQsv4oZi', lastChance: false },
    {
      id: 'CYgQsv4oZi',
      'price_SE.amount': 5000,
      'price_SE.currency': 'SEK'
    }
  ]
}

Hope you can make that out 🙂 !

the reason for the batch is because when this works properly, I'm gonna start importing them by the thousands 😛

Found the issue. I'm working on a fix, and will be able to patch your cluster once I have it ready.

Awesome! Bug?

Yup, in the logic that reconciles fields of repeating document IDs within the same batch.

Okay, glad we found it! Will it be available in the v0.22 or how is your process around fixes of that sort?

Yes, it will be available as part of the 0.22 release, which is not GA yet. Our release cycles are generally longish (few months) because we want to be absolutely sure that GA release is stable.

We balance that by making release candidate builds available to customers that we work closely with on new features, so that they help in both validating the feature and in overall maturity (since they will also be progressing from a dev -> prod for a given new feature that we add).

Sounds great!

I've fixed it locally, along with a test. Thanks for catching this! We will be testing this build further and once it's ready, I will update your cluster with this version and post on this thread.

Awesome, thanks!

We've upgraded your cluster. Please try it out and let me know Dui.

Thank you! It's working now 🙂

One question that is related though:
If I run three upsert in a sequence which are targeting the same document (same id), it doesn't get updated for each upsert. Any ideas?

Upserts of an ID that already exists in the collection, or an upsert where the ID is inserted for the first time with 3 entries in the upsert batch?

basically this payload:

[
  {
    "id": "CYgQsv4oZi",
    "createdAt": 1637160199,
    "updatedAt": 1637576744,
    "metadata.brand": "Nike"
  },
  {
    "id": "CYgQsv4oZi",
    "lastChance": false
  },
  {
    "id": "CYgQsv4oZi",
    "pricing.amount": 50,
    "pricing.currency": "SEK"
  }
]

The end-result is that only the last object in the array gets written to the document.

Does CYgQsv4oZi already exist in the collection before this payload is sent?

No, it does not

Upsert replaces the whole document. If you want to do partial updates, you have use update action.

ah! Okay, I thought upsert was an insert OR update depending on wether the document existed. Like algolia's partialUpdate with createIfNotExists . Does something like that exist?

No, we don't have an upsert+update behavior. This is how it worked in 0.20 but we had to change it because that's not a common behavior for word upsert for people familiar with other DBs so we had to switch to that.

Understandable. Will there be a fourth option eventually, you think?

Yes, I was thinking that maybe an emplace action could be introduced that does upsert if document is not available or does update if it already exists and to which you can either pass whole or partial document.

Nice! Do you know (approximately) in which release?

Can you please create an issue on Github for that for tracking? It will have to be prioritized for the 0.23 (0.22 will be out soon but it's under code freeze and we are only fixing bugs that show up in the final stretch).

Sure, thanks a bunch!

But we can make a 0.23 RC build available as soon as we have feature ready.

There it is 🙂 https://github.com/typesense/typesense/issues/447

Thank you, I will keep you posted!

Dui How's your exploration going? Any other feedback?

Hey! It's going well, thanks for asking!

I'm discussing with my team so we will reach out soon 🙂

Glad to hear! We just released v0.22 GA and the bug you reported earlier is part of that. Thank you!

Glad to hear that 🙂

Would Jan first week be something that would work for you?

No that is perfect!

Thank you, I will update this thread when it's ready to preview.

Sounds great! Another one we've been discussing is the nested objects (https://github.com/typesense/typesense/issues/227), do you know in what release you'll have it in?

That's a lot more intricate change. We will probably be able to work on it only in late Jan or early Feb.

Awesome 🙂 But great to hear that the emplace will happen so soon!

I'll reach out soon and see if we can book a meeting with some more people from my team - we are still investigating if we should move from algolia.

👍 Will be happy to offer balanced inputs.

👋 Dui The emplace action has been implemented. Do you want to do an initial test using a Docker build for initial verification before we upgrade your Cloud cluster?

Wow awesome!!

Yes we can do that.

Awesome, I'll reach out to you soon then 🙂